Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Asynchronous magic with Python

Kevin
August 12, 2015

Asynchronous magic with Python

Learn how to apply Python 3.4's new asyncio module, through the use of generators and select loops.

Kevin

August 12, 2015
Tweet

More Decks by Kevin

Other Decks in Programming

Transcript

  1. • Polyglot developer, currently using Erlang, Elixir, Go, Python, Ruby

    and a bit of Clojure • Been writing Python since 1998, version 1.5 • Writing software professionally for over 25 years About me
  2. Simplest socket server ever import socket server = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)

    server.bind(('127.0.0.1', 8000)) server.listen(1) client, addr = server.accept() while True: data = client.recv(4096) if not data: break print(data) server.close()
  3. Let’s make that better import socket server = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)

    server.bind(('127.0.0.1', 8000) server.listen(8) while True: client, addr = server.accept() while True: data = client.recv(4096) if not data: break print(data) if data == b'CLOSE\r\n': client.close() break server.close()
  4. More than one client at a time? • Forked workers

    • Threaded workers • Select (epoll, kqueue) • Asynchronous workers with asyncio
  5. import socket import os def accept_conn(message, s): while True: client,

    addr = s.accept() print('Got connection while in %s' % message) client.send(bytes('You have connected to %s\n' % message, encoding='utf-8')) while True: data = client.recv(4096) if not data: break print(data) client.close() serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) serversocket.bind(('127.0.0.1', 9000)) serversocket.listen(0) if os.fork() == 0: accept_conn('child', serversocket) accept_conn('parent', serversocket) Forked workers
  6. Forking a child process to listen if os.fork() == 0:

    accept_conn('child', serversocket) accept_conn('parent', serversocket)
  7. import os, socket, threading def accept_conn(client, addr): ident = threading.get_ident()

    print('Got connection while in %s' % ident) client.send(bytes('You have connected to %s\n' % ident, encoding='utf-8')) while True: data = client.recv(4096) if not data: print('Thread %s ending' % (ident)) break print('Thread %s received %s' % (ident, data)) client.close() serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) serversocket.bind(('127.0.0.1', 9000)) serversocket.listen(2) while True: client, addr = serversocket.accept() threading.Thread(None, accept_conn, args=(client, addr), daemon=True).start() Threaded workers
  8. Threaded workers def accept_conn(client, addr): ident = threading.get_ident() print('Got connection

    while in %s' % ident) client.send(bytes('You have connected to %s\n' % ident, encoding='utf-8')) while True: data = client.recv(4096) if not data: print('Thread %s ending' % (ident)) break print('Thread %s received %s' % (ident, data)) client.close()
  9. Threaded workers in stdlib import socketserver import threading class DemoHandler(socketserver.BaseRequestHandler):

    def handle(self): ident = threading.get_ident() print('Got connection while in %s' % ident) self.request.sendall(bytes('You have connected to %s\n' % ident, encoding='utf-8')) while True: data = self.request.recv(4096) if not data: break print(data) class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer): pass server = ThreadedTCPServer(('127.0.0.1', 9001), DemoHandler) server.serve_forever()
  10. import socket, select conns = [] server = socket.socket(socket.AF_INET, socket.

    SOCK_STREAM) server.bind(('127.0.0.1', 7000)) server.listen(10) conns.append(server) print('Listening on 127.0.0.1 7000') while True: rs, ws, es = select.select(conns, [], [], 20) for sock in rs: Select if sock == server: client, addr = server.accept() conns.append(client) print('Client connection received') else: try: data = sock.recv(4096) if not data or data == 'CLOSE\r\n': sock.close() conns.remove(sock) continue else: print(data) except IOError: print('Client %s disconnected' % addr) sock.close() conns.remove(sock) continue server.close()
  11. Select select.select(rlist, wlist, xlist[, timeout]) This is a straightforward interface

    to the Unix select() system call. The first three arguments are sequences of ‘waitable objects’: either integers representing file descriptors or objects with a parameterless method named fileno() returning such an integer: • rlist: wait until ready for reading • wlist: wait until ready for writing • xlist: wait for an “exceptional condition” (see the manual page for what your system considers such a condition) Empty sequences are allowed, but acceptance of three empty sequences is platform-dependent. (It is known to work on Unix but not on Windows.) The optional timeout argument specifies a time-out as a floating point number in seconds. When the timeout argument is omitted the function blocks until at least one file descriptor is ready. A time-out value of zero specifies a poll and never blocks. The return value is a triple of lists of objects that are ready: subsets of the first three arguments. When the time-out is reached without a file descriptor becoming ready, three empty lists are returned. rs, ws, es = select.select(conns, [], [], 20)
  12. socket Select polling application select() register new client process data

    socket client sockets server socket new connection received data timeout remove client no data 127.0.0.1, 7000 server socket is readable? client socket is readable? call recv()
  13. selectors • More efficient versions of select ◦ epoll -

    Linux only ◦ kqueue - BSDs only (includes Darwin) • selectors module uses the best version for the host platform
  14. PEP 3156 • December 2012 • Based on PEP 3153

    which didn’t get approved • “This is a proposal for asynchronous I/O in Python 3, starting at Python 3.3…[the] proposal includes a pluggable event loop, transport and protocol abstractions similar to those in Twisted, and a higher- level scheduler based on yield from (PEP 380).”
  15. Call later import asyncio def hello_world(loop): print('Hello World', loop) loop.stop()

    loop = asyncio.get_event_loop() () loop.call_later(20, hello_world, loop) loop.run_forever() loop.close()
  16. Call later import asyncio def hello_world(loop): print('Hello World', loop) def

    hello_world2(loop): print('Hello World2', loop) loop.stop() loop = asyncio.get_event_loop() handle = loop.call_later(18, hello_world, loop) loop.call_later(20, hello_world2, loop) loop.run_forever() loop.close()
  17. import inspect def testing(y): while y > 0: yield y

    y = y -1 x = testing(5) print(inspect.getgeneratorstate(x)) print(inspect.getgeneratorlocals(x)) print(next(x)) print(inspect.getgeneratorlocals(x)) print(next(x)) print(inspect.getgeneratorlocals(x)) print(next(x)) Inspecting generator state
  18. yield from - delegate to a generator def testing3(): yield

    10 yield 20 def testing2(): yield 100 yield from testing3() yield 200 def testing1(): yield 100 yield testing3() yield 200 print([x for x in testing1()]) print([x for x in testing2()])
  19. Generators import asyncio import inspect @asyncio.coroutine def do_the_thing(y): print('About to

    sleep') yield from asyncio.sleep(5) print('Finished sleeping') return y @asyncio.coroutine def testing(y): return (yield from do_the_thing(y)) loop = asyncio.get_event_loop() print(loop.run_until_complete(testing(5)))
  20. EventLoop Task @asyncio.coroutine testing(5) do_the_thing(5) yield from do_the_thing(5) print('About to

    sleep') yield from asyncio.sleep(5) print('Finished sleeping') return y return (yield from do_the_thing(y)) select timeout 5s
  21. A future is…”an object that acts as a proxy for

    a result that is initially unknown, usually because the computation of its value is yet incomplete.” Futures
  22. Python futures import asyncio @asyncio.coroutine def slow_operation(future): yield from asyncio.sleep(1)

    future.set_result(2000) def got_result(future): print(future.result()) loop.stop() future = asyncio.Future() asyncio.Task(slow_operation(future)) future.add_done_callback(got_result) loop = asyncio.get_event_loop() loop.run_forever() loop.close()
  23. Concurrent tasks import asyncio, aiohttp @asyncio.coroutine def do_the_thing(y): print('About to

    sleep') yield from asyncio.sleep(5) print('Finished sleeping') return y @asyncio.coroutine def testing(y): return (yield from do_the_thing(y)) @asyncio.coroutine def get_body(url): print('About to fetch') response = yield from aiohttp.request('GET', url) print('About to read content') return (yield from response.read()) loop = asyncio.get_event_loop() print(loop.run_until_complete(asyncio.gather( testing(5), get_body('http://bigkevmcd.com/demo.txt'))))
  24. Asynchronous HTTP @asyncio.coroutine def get_body(url): print('About to fetch') response =

    yield from aiohttp.request('GET', url) print('About to read content') return (yield from response.read())
  25. Concurrent HTTP requests import asyncio import aiohttp @asyncio.coroutine def get_body(url):

    response = (yield from asyncio.wait_for(aiohttp.request('GET', url), 10)) return (yield from response.read()) urls_to_fetch = [ 'http://bigkevmcd.com/demo.txt', 'http://bigkevmcd.com/demo.txt', 'http://bigkevmcd.com/demo.txt' ] loop = asyncio.get_event_loop() tasks = asyncio.gather(*[get_body(url) for url in urls_to_fetch]) print(tasks) result = loop.run_until_complete(tasks) print(result)
  26. Asynchronous HTTP Services import os import asyncio from aiohttp import

    web @asyncio.coroutine def hello(request): return web.Response(body='Hello, world!'.encode('utf-8'), content_type='text/plain') app = web.Application() app.router.add_route('GET', '/', hello)
  27. Asynchronous wsgi handling import asyncio import aiohttp def get_http_body(url): print('Starting

    GET') response = yield from aiohttp.request('GET', url) print('Reading body') return (yield from response.read()) def app(environ, start_response): data = (yield from get_http_body('http://bigkevmcd.com'))[:20] start_response("200 OK", [ ("Content-Type", "text/plain"), ("Content-Length", str(len(data))) ]) return iter([data])
  28. import os import asyncio import asyncio_redis from aiohttp import web

    @asyncio.coroutine def root(request): return web.HTTPFound('/static/websockets1.html') app = web.Application() app.router.add_route('GET', '/', root) app.router.add_route('GET', '/redis', redis_handler) app.router.add_static('/static', os.path.join(os.path. dirname(__file__), 'static')) Driving websockets with Redis @asyncio.coroutine def redis_handler(request): connection = yield from asyncio_redis.Connection.create( host='localhost', port=6379) subscriber = yield from connection.start_subscribe() yield from subscriber.subscribe(['messages']) resp = web.WebSocketResponse() resp.start(request) while True: reply = yield from subscriber.next_published() resp.send_str(reply.value) connection.close()
  29. Simple protocol import asyncio class DemoProtocol(asyncio.Protocol): def connection_made(self, transport): print('Received

    a connection') self.transport = transport def data_received(self, data): message = data.decode().strip() print('Data received:', message) if data == 'CLOSE: print('Closing connection') self.transport.close() loop = asyncio.get_event_loop() factory = loop.create_server( DemoProtocol, '127.0.0.1', 8888) server = loop.run_until_complete(factory) print('Listening on 127.0.0.1 8888') try: loop.run_forever() except KeyboardInterrupt: pass server.close() loop.run_until_complete(server.wait_closed()) loop.close()
  30. When not to use asyncio? • Asynchronous IO does not

    magically use more CPUs ◦ if processing the data is CPU intensive, then you will want to distribute that load over multiple CPUs ◦ asyncio has support for handling callbacks within threads, but this has the same kinds of issues as threading on Python normally has.
  31. Asyncio addons • http://asyncio.org/ ◦ Go here for asyncio addons

    • aiopg - asynchronous PostgreSQL querying ◦ https://github.com/aio-libs/aiopg • AsyncSSH ◦ http://asyncssh.readthedocs.org/en/latest/ • aioredis ◦ http://asyncio-redis.readthedocs.org/en/latest/ • AioDocker ◦ https://github.com/paultag/aiodocker ◦ Asynchronously subscribe to events from Docker
  32. Asyncssh import sys import asyncio import aiohttp import asyncssh @asyncio.coroutine

    def get_http_body(url): print('Fetching HTTP file') response = yield from aiohttp.request('GET', url) print('Got HTTP response') return (yield from response.read()) @asyncio.coroutine def fetch_file_from_sftp(hostname, username, filename): with (yield from asyncssh.connect(hostname, username=username)) as conn: print('Connected to SSH server') with (yield from conn.start_sftp_client()) as sftp: print('Retrieving file') yield from sftp.get(filename)
  33. Fetching files via ssh loop = asyncio.get_event_loop() print(loop.run_until_complete(asyncio.gather( fetch_file_from_sftp('example.com', 'testing',

    '/srv/www/example.com/www/demo.txt'), fetch_file_from_sftp('example.com', 'testing', 'kevin.txt'), get_http_body('http://example.com/demo.txt'))))