DefaultSelector automatically chooses the most efficient available operating-system selector, keeping selector-based socket code portable across systems. section 3.4
Use it as the coordination point for a non-blocking socket event loop:
- Create a DefaultSelector and register sockets with the read or write events that matter. section 3.4
- Call select with an optional timeout; it waits efficiently and returns sockets ready for processing. section 3.4
- Process connections and client data, then register or deregister sockets as their monitoring needs change. section 3.4
Listing 3.7 Using selectors to build a non-blocking server
import selectors
import socket
from selectors import SelectorKey
from typing import List, Tuple
selector = selectors.DefaultSelector()
server_socket = socket.socket()
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_address = ('127.0.0.1', 8000)
server_socket.setblocking(False)
server_socket.bind(server_address)
server_socket.listen()
selector.register(server_socket, selectors.EVENT_READ)
while True:
events: List[Tuple[SelectorKey, int]] = selector.select(timeout=1) #1
if len(events) == 0: #2
print('No events, waiting a bit more!')
for event, _ in events:
event_socket = event.fileobj #3
if event_socket == server_socket: #4
connection, address = server_socket.accept()
connection.setblocking(False)
print(f"I got a connection from {address}")
selector.register(connection, selectors.EVENT_READ) #5
else:
data = event_socket.recv(1024) #6
print(f"I got some data: {data}")
event_socket.send(data)
Builds a non-blocking echo server with DefaultSelector, a one-second timeout, server-socket registration, and client-event handling. section 3.4
