Debugging blocking calls tip

import asyncio
import time

loop = asyncio.get_event_loop()
loop.slow_callback_duration = 0.01

async def sleeper():
    time.sleep(0.1)  # we block here

loop.run_until_complete(sleeper())
					

Executing <Task finished coro=<sleeper() done, defined at
code/debug_example.py:9> result=None created at
/usr/local/lib/python3.5/asyncio/base_events.py:323>
took 0.102 seconds
					

Third Party Libraries Pro Tip

@asyncio.coroutine
def get_async(self):
    """Sends the GET request using an asyncio coroutine
    .... 
    """
    future = self._client._loop.run_in_executor(None,
                                                self.get)
    collection_response = yield from future
    return collection_response
					

Approach #2 Is REST API available?

• DynamoDB
• Neo4j
• Elasticsearch
• HBase
• HDFS

• CouchDB
• Riak
• VoltDB
• InfluxDB
• ArangoDB

REST Client Tip

import asyncio
import aiohttp

# carry the loop Luke!
loop = asyncio.get_event_loop()

async def go():
    session = aiohttp.ClientSession(loop=loop)
    async with session.get('http://python.org') as resp:
        data = await resp.text()
        print(data)
    session.close()

loop.run_until_complete(go())
					

Approach #3 Is there simple text or binary protocol?

Example of databases and message queues with binary protocol:

• redis
• memcached
• couchebase

• grearman
• beanstalkd
• disque

Do not afraid to get your hands dirty.

import asyncio, struct
from aiogibson import encode_command

async def read_from_connection(host, port, *, loop=None):
    reader, writer = await asyncio.open_connection(
    host, port, loop=loop)

    cmd = encode_command(b'GET', 'key')
    writer.write(cmd)
    
    header = await reader.readexactly(4 + 2 + 1)
    unpacked = struct.unpack(b'<HBI', header)
    code, gb_encoding, resp_size = unpacked
    payload = await reader.readexactly(resp_size)
    print(payload)
					

Protocol Pipelining

def execute(self):
    cmd = encode_command(b'GET', 'key')
    self.writer.write(cmd)
    fut = asyncio.Future(loop=self._loop)
    self._queue.append(fut)
    return fut

async def reader_task(self):
    while True:
        header = await self.reader.readexactly(4 + 2 + 1)
        unpacked = struct.unpack(b'<HBI', header)
        code, gb_encoding, resp_size = unpacked
        # wait and read payload
        payload = await reader.readexactly(resp_size)
        future = self._queue.pop()
        future.set_result(payload)
					

Approach #4 Is Sync Python Client available?

1. Locate socket.recv()
2. Replace with await reader.read()
3. Make function coroutine with async def
4. Call this function with await
5. Call parent functions with await

Approach #5 Is there universal solution to all problems?

import asyncio
from pyodbc import connect

loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=4)
async def test_example():
    dsn = 'Driver=SQLite;Database=sqlite.db'
    conn = await loop.run_in_executor(executor, connect, dsn)
    cursor = await loop.run_in_executor(executor, conn.cursor)
    conn = await loop.run_in_executor(executor, cursor.execute,
                                      'SELECT 42;')

loop.run_until_complete(test_example())
					

But how I know which method to call in thread?

requests.get()
					

with nogil:
    [code to be executed with the GIL released]
					

For C extension

Py_BEGIN_ALLOW_THREADS
ret = SQLDriverConnect(hdbc, 0, szConnect, SQL_NTS,
                       0, 0, 0, SQL_DRIVER_NOPROMPT);
Py_END_ALLOW_THREADS

					

What about FileSystem IO?

asyncio does not support asynchronous operations on the filesystem due to OS limitations.

aiofiles library workaround

async def go():
    f = await aiofiles.open('filename', mode='r')
    try:
        data = await f.read()
    finally:
        await f.close()
    print(data)
loop.run_until_complete(go())
					

What about CPU intensive task?

loop = asyncio.get_event_loop()
executor = ProcessPoolExecutor(max_workers=3)

def is_prime(n):
    if n % 2 == 0: return False
    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0: return False
    return True

async def go():
    n = 112272535095293
    result = await loop.run_in_executor(executor, is_prime, n)
loop.run_until_complete(go(loop, executor))

					

Thanks!