python 来自我的PyCon Brasil 2015主题演讲的实时编码示例

Posted 2021-05-09

# play.py

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

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

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

    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:
                 nsent = sock.send(data)
                 data = data[nsent:]
             except BlockingIOError:
                 await write_wait(sock)

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

    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)  # Run to the yield
                    getattr(self, op)(*args)  # Sneaky method call
                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)

# pecho.py

from socket import *
import play

loop = play.Loop()

async def echo_server(address):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(address)
    sock.listen(5)
    sock.setblocking(False)
    while True:
        client, addr = await loop.sock_accept(sock)
        print('Connection from', addr)
        loop.create_task(echo_handler(client))

async def echo_handler(client):
    with client:
        while True:
            data = await loop.sock_recv(client, 10000)
            if not data:
                break
            await loop.sock_sendall(client, b'Got:'+data)
    print('Connection closed')

loop.create_task(echo_server(('',25000)))
loop.run_forever()

from socket import *
import time

def benchmark(addr, nmessages):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.connect(addr)
    start = time.time()
    for n in range(nmessages):
        sock.send(b'x')
        resp = sock.recv(10000)
    end = time.time()
    print(nmessages/(end-start), 'messages/sec')

benchmark(('localhost',25000), 100000)

# aecho.py

from socket import *
import asyncio

loop = asyncio.get_event_loop()

async def echo_server(address):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(address)
    sock.listen(5)
    sock.setblocking(False)
    while True:
        client, addr = await loop.sock_accept(sock)
        print('Connection from', addr)
        loop.create_task(echo_handler(client))

async def echo_handler(client):
    with client:
        while True:
            data = await loop.sock_recv(client, 10000)
            if not data:
                break
            await loop.sock_sendall(client, b'Got:'+data)
    print('Connection closed')

loop.create_task(echo_server(('',25000)))
loop.run_forever()