Python:EventLoop
lastmod 2024-07-17created 2024-07-171 edits
The event loop used by asyncio.
Python Concurrency Mind Maps
| |
Executing code in thread or process pools
awaitable loop.run_in_executor(executor, func, *args)
Arrange for func to be called in the specified executor.
loop.set_default_executor(executor)
Set executor as the default executor used by run_in_executor(). executor must be an instance of ThreadPoolExecutor.
Watching file descriptors
See also the Platform Support section for some limitations of these methods.
loop.add_reader(fd, callback, *args)
Start monitoring the fd file descriptor for read availability and invoke callback with the specified arguments once fd is available for reading.
Here is an example:
import asyncio
import urllib.parse
import sys
import pdb
import os
def fileCallback(*args):
print("Received: " + sys.stdin.readline())
loop = asyncio.get_event_loop()
task = loop.add_reader(sys.stdin.fileno(), fileCallback)
loop.run_forever()
loop.remove_reader(fd)
Stop monitoring the fd file descriptor for read availability. Returns True if fd was previously being monitored for reads.
loop.add_writer(fd, callback, *args)
Start monitoring the fd file descriptor for write availability and invoke callback with the specified arguments once fd is available for writing.
Use functools.partial() to pass keyword arguments to callback.
It looks something like this.
loop.remove_writer(fd)
Stop monitoring the fd file descriptor for write availability. Returns True if fd was previously being monitored for writes.
How to await a select.select call in python asyncio
You could use #loop.add_reader to wait for the read availability of your socket:
async def watch(fd):
future = asyncio.Future()
loop.add_reader(fd, future.set_result, None)
future.add_done_callback(lambda f: loop.remove_reader(fd))
await future
async def run(self):
while True:
await watch(self._q._connection)
msg = self._q.receive()
print(msg)
However, it'll be very tricky to avoid all the blocking IO calls of the library you mentioned without rewriting it completely. Instead, I'd recommend to use the #loop.run_in_executor method to schedule the blocking IO calls in a thread pool:
async def run(self):
loop = asyncio.get_event_loop()
while True:
msg = await loop.run_in_executor(None, self._q.receive)
print(msg)
If you want to set up I/O multiplexing with Python:selectors
import asyncio
import selectors
selector = selectors.SelectSelector()
loop = asyncio.SelectorEventLoop(selector)
asyncio.set_event_loop(loop)