Asynchronous Server Basic Example

async_server.py

"""
The server code is based entirely on D. Beazley's talk "Topics of Interest (Python Asyncio)"
given in 2015 at the Python Brazil conference.

http://pyvideo.org/python-brasil-2015/keynote-david-beazley-topics-of-interest-python-asyncio.html

The server can handle 60,000 requests in 2.020 seconds (29,696 requests/second) in a ~2015 macOS.

"""

from socket import *
from types import coroutine
from collections import deque
from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE

@coroutine
def read_wait(sock):
    yield ('read_wait', sock)

async def write_wait(sock):
    yield ('write_wait', sock)

class Loop:

    def __init__(self):
        self.ready = deque()
        self.selector = DefaultSelector()

    def run_forever(self):
        while True:
            while not self.ready:
                events = self.selector.select()
                for key, _ in events:
                    self.ready.append(key.data)
                    self.selector.unregister(key.fileobj)

            while self.ready:
                self.current_task = self.ready.popleft()
                try:
                    op, *args = self.current_task.send(None)
                    getattr(self, op)(*args)
                except StopIteration:
                    pass

    def read_wait(self, sock):
        self.selector.register(sock, EVENT_READ, self.current_task)

    def write_wait(self, sock):
        self.selector.register(sock, EVENT_WRITE, self.current_task)

    def create_task(self, coro):
        self.ready.append(coro)

    async def sock_recv(self, sock, maxbytes):
        await read_wait(sock)
        return sock.recv(maxbytes)

    async def sock_accept(self, sock):
        await read_wait(sock)
        return sock.accept()

    async def sock_sendall(self, sock, data):
        while data:
            try:
                num_sent = sock.send(data)
                data = data[num_sent:]
            except BlockingIOError:
                await write_wait(sock)

loop = Loop()

async def server(address):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(address)
    sock.listen()
    sock.setblocking(False)
    while True:
        client, addr = await loop.sock_accept(sock)
        print(f"connection: {addr}")
        loop.create_task(handler(client))

RESPONSE = """\
HTTP/1.1 200 OK\r
content-length: 0\r
connection: keep-alive\r
server: simple server\r
\r
""".encode("utf-8")
def headers(data):
    beg = data.find("\r\n") + 2
    end = data.find("\r\n\r\n")
    return { k.lower().strip() : v.strip() for k, v in
             (line.split(":", 1) for line in
              data[beg:end].split("\r\n")) }

async def handler(client):
    with client:
        while True:
            data = await loop.sock_recv(client, 1024)
            if not data:
                break
            h = headers(data.decode("utf-8"))
            await loop.sock_sendall(client, RESPONSE)
    print("connection closed")

loop.create_task(server(('', 8000)))
loop.run_forever()

benchmark.py

import time
import threading
from socket import *

REQUEST = b"""\
GET / HTTP/1.1\r
connection: keep-alive\r
user-agent: benchmark\r
\r
"""

def request(N):
    """Each "request" makes `N` new connections, and each
    connection sends `10 * N` messages."""
    for _ in range(N):
        sock = socket(AF_INET, SOCK_STREAM)
        sock.connect(("", 8000))
        for _ in range(10 * N):
            sock.sendall(REQUEST)
            data = sock.recv(1024)
        sock.close()

def benchmark(N, M):
    threads = []
    tbeg = time.perf_counter()
    for _ in range(M): # run the "request" M times on different threads
        thread = threading.Thread(target=request, args=(N,))
        threads.append(thread)
        thread.start()
    for thread in threads:
        thread.join()
    tend = time.perf_counter()
    total = 10 * M * N**2
    elapsed = tend - tbeg
    rate = total / elapsed
    print(f"{total:,.0f} requests in {elapsed:.3f} seconds",
          f"({rate:,.0f} requests/second)")

if __name__ == "__main__":
    benchmark(20, 15)