A Hitchhikers Guide To Asynchronous Programming — Pysheeet

A Hitchhikers Guide to Asynchronous Programming¶

Table of Contents

• A Hitchhikers Guide to Asynchronous Programming

  • Abstract

  • Introduction

  • Callback Functions

  • Event Loop

  • What is a Coroutine?

  • Conclusion

  • Reference

Abstract¶

The C10k problem is still a puzzle for a programmer to find a way to solve it. Generally, developers deal with extensive I/O operations via thread, epoll, or kqueue to avoid their software waiting for an expensive task. However, developing a readable and bug-free concurrent code is challenging due to data sharing and job dependency. Even though some powerful tools, such as Valgrind, help developers to detect deadlock or other asynchronous issues, solving these problems may be time-consuming when the scale of software grows large. Therefore, many programming languages such as Python, Javascript, or C++ dedicated to developing better libraries, frameworks, or syntaxes to assist programmers in managing concurrent jobs properly. Instead of focusing on how to use modern parallel APIs, this article mainly concentrates on the design philosophy behind asynchronous programming patterns.

Using threads is a more natural way for developers to dispatch tasks without blocking the main thread. However, threads may lead to performance issues such as locking critical sections to do some atomic operations. Although using event-loop can enhance performance in some cases, writing readable code is challenging due to callback problems (e.g., callback hell). Fortunately, programming languages like Python introduced a concept, async/await, to help developers write understandable code with high performance. The following figure shows the main goal by using async/await to handle socket connections like utilizing threads.

Introduction¶

Handling I/O operations such as network connections is one of the most expensive tasks in a program. Take a simple TCP blocking echo server as an example (The following snippet). If a client connects to the server successfully without sending any request, it blocks others’ connections. Even though clients send data as soon as possible, the server cannot handle other requests if there is no client trying to establish a connection. Also, handling multiple requests is inefficient because it wastes a lot of time waiting for I/O responses from hardware such as network interfaces. Thus, socket programming with concurrency becomes inevitable to manage extensive requests.

import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("127.0.0.1", 5566)) s.listen(10) while True: conn, addr = s.accept() msg = conn.recv(1024) conn.send(msg)

One possible solution to prevent a server waiting for I/O operations is to dispatch tasks to other threads. The following example shows how to create a thread to handle connections simultaneously. However, creating numerous threads may consume all computing power without high throughput. Even worse, an application may waste time waiting for a lock to process tasks in critical sections. Although using threads can solve blocking issues for a socket server, other factors, such as CPU utilization, are essential for a programmer to overcome the C10k problem. Therefore, without creating unlimited threads, the event loop is another solution to manage connections.

import threading import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("127.0.0.1", 5566)) s.listen(10240) def handler(conn): while True: msg = conn.recv(65535) conn.send(msg) while True: conn, addr = s.accept() t = threading.Thread(target=handler, args=(conn,)) t.start()

A simple event-driven socket server includes three main components: an I/O multiplexing module (e.g., select), a scheduler (loop), and callback functions (events). For example, the following server utilizes the high-level I/O multiplexing, selectors, within a loop to check whether an I/O operation is ready or not. If data is available to read/write, the loop acquires I/O events and execute callback functions, accept, read, or write, to finish tasks.

import socket from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE from functools import partial s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("127.0.0.1", 5566)) s.listen(10240) s.setblocking(False) sel = DefaultSelector() def accept(s, mask): conn, addr = s.accept() conn.setblocking(False) sel.register(conn, EVENT_READ, read) def read(conn, mask): msg = conn.recv(65535) if not msg: sel.unregister(conn) return conn.close() sel.modify(conn, EVENT_WRITE, partial(write, msg=msg)) def write(conn, mask, msg=None): if msg: conn.send(msg) sel.modify(conn, EVENT_READ, read) sel.register(s, EVENT_READ, accept) while True: events = sel.select() for e, m in events: cb = e.data cb(e.fileobj, m)

Although managing connections via threads may not be efficient, a program that utilizes an event loop to schedule tasks isn’t easy to read. To enhance code readability, many programming languages, including Python, introduce abstract concepts such as coroutine, future, or async/await to handle I/O multiplexing. To better understand programming jargon and using them correctly, the following sections discuss what these concepts are and what kind of problems they try to solve.

Callback Functions¶

A callback function is used to control data flow at runtime when an event is invoked. However, preserving current callback function’s status is challenging. For example, if a programmer wants to implement a handshake over a TCP server, he/she may require to store previous status in some where.

import socket from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE from functools import partial s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("127.0.0.1", 5566)) s.listen(10240) s.setblocking(False) sel = DefaultSelector() is_hello = {} def accept(s, mask): conn, addr = s.accept() conn.setblocking(False) is_hello[conn] = False; sel.register(conn, EVENT_READ, read) def read(conn, mask): msg = conn.recv(65535) if not msg: sel.unregister(conn) return conn.close() # check whether handshake is successful or not if is_hello[conn]: sel.modify(conn, EVENT_WRITE, partial(write, msg=msg)) return # do a handshake if msg.decode("utf-8").strip() != "hello": sel.unregister(conn) return conn.close() is_hello[conn] = True def write(conn, mask, msg=None): if msg: conn.send(msg) sel.modify(conn, EVENT_READ, read) sel.register(s, EVENT_READ, accept) while True: events = sel.select() for e, m in events: cb = e.data cb(e.fileobj, m)

Although the variable is_hello assists in storing status to check whether a handshake is successful or not, the code becomes harder for a programmer to understand. In fact, the concept of the previous implementation is simple. It is equal to the following snippet (blocking version).

def accept(s): conn, addr = s.accept() success = handshake(conn) if not success: conn.close() def handshake(conn): data = conn.recv(65535) if not data: return False if data.decode('utf-8').strip() != "hello": return False conn.send(b"hello") return True

To migrate the similar structure from blocking to non-blocking, a function (or a task) requires to snapshot the current status, including arguments, variables, and breakpoints, when it needs to wait for I/O operations. Also, the scheduler should be able to re-entry the function and execute the remaining code after I/O operations finish. Unlike other programming languages such as C++, Python can achieve the concepts discussed above easily because its generator can preserve all status and re-entry by calling the built-in function next(). By utilizing generators, handling I/O operations like the previous snippet but a non-blocking form, which is called inline callback, is reachable inside an event loop.

Event Loop¶

An event loop is a scheduler to manage tasks within a program instead of depending on operating systems. The following snippet shows how a simple event loop to handle socket connections asynchronously. The implementation concept is to append tasks into a FIFO job queue and register a selector when I/O operations are not ready. Also, a generator preserves the status of a task that allows it to be able to execute its remaining jobs without callback functions when I/O results are available. By observing how an event loop works, therefore, it would assist in understanding a Python generator is indeed a form of coroutine.

# loop.py from selectors import DefaultSelector, EVENT_READ, EVENT_WRITE class Loop(object): def __init__(self): self.sel = DefaultSelector() self.queue = [] def create_task(self, task): self.queue.append(task) def polling(self): for e, m in self.sel.select(0): self.queue.append((e.data, None)) self.sel.unregister(e.fileobj) def is_registered(self, fileobj): try: self.sel.get_key(fileobj) except KeyError: return False return True def register(self, t, data): if not data: return False if data[0] == EVENT_READ: if self.is_registered(data[1]): self.sel.modify(data[1], EVENT_READ, t) else: self.sel.register(data[1], EVENT_READ, t) elif data[0] == EVENT_WRITE: if self.is_registered(data[1]): self.sel.modify(data[1], EVENT_WRITE, t) else: self.sel.register(data[1], EVENT_WRITE, t) else: return False return True def accept(self, s): conn, addr = None, None while True: try: conn, addr = s.accept() except BlockingIOError: yield (EVENT_READ, s) else: break return conn, addr def recv(self, conn, size): msg = None while True: try: msg = conn.recv(1024) except BlockingIOError: yield (EVENT_READ, conn) else: break return msg def send(self, conn, msg): size = 0 while True: try: size = conn.send(msg) except BlockingIOError: yield (EVENT_WRITE, conn) else: break return size def once(self): self.polling() unfinished = [] for t, data in self.queue: try: data = t.send(data) except StopIteration: continue if self.register(t, data): unfinished.append((t, None)) self.queue = unfinished def run(self): while self.queue or self.sel.get_map(): self.once()

By assigning jobs into an event loop to handle connections, the programming pattern is similar to using threads to manage I/O operations but utilizing a user-level scheduler. Also, PEP 380 enables a generator delegation, which allows a generator can wait for other generators to finish their jobs. Obviously, the following snippet is more intuitive and readable than using callback functions to handle I/O operations.

# foo.py # $ python3 foo.py & # $ nc localhost 5566 import socket from selectors import EVENT_READ, EVENT_WRITE # import loop.py from loop import Loop s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("127.0.0.1", 5566)) s.listen(10240) s.setblocking(False) loop = Loop() def handler(conn): while True: msg = yield from loop.recv(conn, 1024) if not msg: conn.close() break yield from loop.send(conn, msg) def main(): while True: conn, addr = yield from loop.accept(s) conn.setblocking(False) loop.create_task((handler(conn), None)) loop.create_task((main(), None)) loop.run()

Using an event loop with syntax, yield from, can manage connections without blocking the main thread, which is the usage of the module, asyncio, before Python 3.5. However, using the syntax, yield from, is ambiguous because it may tie programmers in knots: why adding @asyncio.coroutine makes a generator become a coroutine? Instead of using yield from to handle asynchronous operations, PEP 492 proposes that coroutine should become a standalone concept in Python, and that is how the new syntax, async/await, was introduced to enhance readability for asynchronous programming.

What is a Coroutine?¶

Python document defines that coroutines are a generalized form of subroutines. However, this definition is ambiguous and impedes developers to understand what coroutines are. Based on the previous discussion, an event loop is responsible for scheduling generators to perform specific tasks, and that is similar to dispatch jobs to threads. In this case, generators serve like threads to be in charge of “routine jobs.” Obviously, A coroutine is a term to represent a task that is scheduled by an event-loop in a program instead of operating systems. The following snippet shows what @coroutine is. This decorator mainly transforms a function into a generator function and using a wrapper, types.coroutine, to preserve backward compatibility.

import asyncio import inspect import types from functools import wraps from asyncio.futures import Future def coroutine(func): """Simple prototype of coroutine""" if inspect.isgeneratorfunction(func): return types.coroutine(func) @wraps(func) def coro(*a, **k): res = func(*a, **k) if isinstance(res, Future) or inspect.isgenerator(res): res = yield from res return res return types.coroutine(coro) @coroutine def foo(): yield from asyncio.sleep(1) print("Hello Foo") loop = asyncio.get_event_loop() loop.run_until_complete(loop.create_task(foo())) loop.close()

Conclusion¶

Asynchronous programming via an event loop becomes more straightforward and readable nowadays due to modern syntaxes and libraries’ support. Most programming languages, including Python, implement libraries to manage task scheduling via interacting with new syntaxes. While new syntaxes look enigmatic in the beginning, they provide a way for programmers to develop logical structure in their code, like using threads. Also, without calling a callback function after a task finish, programmers do not need to worry about how to pass the current task status, such as local variables and arguments, into other callbacks. Thus, programmers will be able to focus on developing their programs without wasting a log of time to troubleshoot concurrent issues.

Reference¶

1. asyncio — Asynchronous I/O

2. PEP 342 - Coroutines via Enhanced Generators

3. PEP 380 - Syntax for Delegating to a Subgenerator

4. PEP 492 - Coroutines with async and await syntax

This project tries to provide many snippets of Python code that make life easier.

Useful Links

• pysheeet website
• pysheeet @ GitHub
• Issue Tracker
• pysheeet as a PDF

Cheat Sheets

• C/C++ cheat sheet

Table of Contents

• A Hitchhikers Guide to Asynchronous Programming
  • Abstract
  • Introduction
  • Callback Functions
  • Event Loop
  • What is a Coroutine?
  • Conclusion
  • Reference

Quick search

©2016-2024, crazyguitar. | Page source