A selector is the event loop’s readiness monitor: it watches registered sockets, waits for read events, and invokes callbacks when processing can continue. section 14.5.6
Within a custom asynchronous event loop, a selector:
- Registers nonblocking sockets for read events and associates each registration with a callback. section 14.5.6
- Modifies an existing socket registration when the same socket is registered again. section 14.5.6
- Wakes the loop when socket activity occurs, allowing futures, tasks, and suspended coroutines to advance. section 14.5.6
Concurrent socket server
A custom server uses the selector to accept connections and monitor each client for incoming data. Multiple Telnet clients can connect and send messages concurrently. section 14.5.7
Listing 14.12 An event loop implementation
import functools
import selectors
from typing import List
from chapter_14.listing_14_11 import CustomTask
from chapter_14.listing_14_8 import CustomFuture
class EventLoop:
_tasks_to_run: List[CustomTask] = []
def __init__(self):
self.selector = selectors.DefaultSelector()
self.current_result = None
def _register_socket_to_read(self, sock, callback): #1
future = CustomFuture()
try:
self.selector.get_key(sock)
except KeyError:
sock.setblocking(False)
self.selector.register(sock, selectors.EVENT_READ, functools.partial(callback, future))
else:
self.selector.modify(sock, selectors.EVENT_READ, functools.partial(callback, future))
return future
def _set_current_result(self, result):
self.current_result = result
async def sock_recv(self, sock): #2
print('Registering socket to listen for data...')
return await self._register_socket_to_read(sock, self.recieved_data)
async def sock_accept(self, sock): #3
print('Registering socket to accept connections...')
return await self._register_socket_to_read(sock, self.accept_connection)
def sock_close(self, sock):
self.selector.unregister(sock)
sock.close()
def register_task(self, task): #4
self._tasks_to_run.append(task)
def recieved_data(self, future, sock):
data = sock.recv(1024)
future.set_result(data)
def accept_connection(self, future, sock):
result = sock.accept()
future.set_result(result)
def run(self, coro): #5
self.current_result = coro.send(None)
while True:
try:
if isinstance(self.current_result, CustomFuture):
self.current_result.add_done_callback(self._set_current_result)
if self.current_result.result() is not None:
self.current_result = coro.send(self.current_result.result())
else:
self.current_result = coro.send(self.current_result)
except StopIteration as si:
return si.value
for task in self._tasks_to_run:
task.step()
self._tasks_to_run = [task for task in self._tasks_to_run if not task.is_finished()]
events = self.selector.select()
print('Selector has an event, processing...')
for key, mask in events:
callback = key.data
callback(key.fileobj)
The implementation shows DefaultSelector setup, nonblocking sockets, read-event registration, and callback attachment. section 14.5.6
