Aiohttp Usage

About Me

• R / Python
• PyCon APAC 2014-16/TW staff
  We're calling for reviewers
• NTU Bioinfo master student
• Former intern at Microsoft Research Asia
• Intern at Pinkoi
• Happy to chat about bioinfo and anime

這幾天在分析 ElasticSearch 結果

• 他可以當成一個 RESTful API，餵個字串就會回傳搜尋結果
• 總之要一直 GET 他
• 因為要做 benchmark，所以要 GET 很多次
• 真的很多次，大概幾十萬次

雖然 ES 很快一個 query 約 15ms，十萬次也要半個小時。耍廢半小時？

所以想到用個非同步打 ES 好了。mutlithreading / async?

        from concurrent.futures import ThreadPoolExecutor

def search(query, url='/search'):
    r = requests.request('GET', url, params={'q': query})
    quote_json = r.json()
    return quote_json

with ThreadPoolExecutor(max_workers) as executor:
    results = executor.map(search, all_queries)
      

Multithreading 不夠潮

• 用起來應該沒問題，官網也是用網路爬蟲當例子
• 不過身為一個（一直出包）不斷實驗改進的工程師，還蠻常活在 REPL / IPython Notebook 裡，印象中 threading、multiprocessing 在這情況可能會噴 error（雖然我沒有測啦）
• 廢話這多，總之想試 async，多點些技能
• async 版的 requests 就是 aiohttp

Event loop / asyncio 是啥？可以吃嗎？

• 可以直接吃，但你會投降 Stop, I yield!（冰火粉覺得好冷）
• 我也是看報紙才知道的。看了
• TP 大大 Yield a Tulip 投影片（影片）
• David Beazley 現場刻一個 (source code)
  Python Concurrency From the Ground Up
• 從 asyncio 大本營看了有的沒的大大小小教學
• …（然後進入精神時光屋差不多幾個月）
• 就會了呢

情境測試：Random Quoter

• 用 Tornado 寫了個 server （用它自己的 event loop）
• http://localhost:5566/quote 就會回傳 1984 隨機段落
  但它反應太快了 ?slow=true 強制加 0.5 秒
• /quote?part=1&chapter=2 可以選段落章節抽樣
• /quote/uniform 確保每個段落機率是一致的（統計強迫症）

python 0_parse_1984.py
python random_quote_app.py [--help|--num_process=1]

用起來大概像這樣

$ curl -s -XGET "http://localhost:5566/quote" | \
  python -m json.tool  # prettify
{
    "quote": "   WAR IS PEACE   FREEDOM IS SLAVERY
                 IGNORANCE IS STRENGTH",
    "source": {
        "chapter": 1,
        "part": 1
    }
}

$ curl -s -XGET "localhost:5566/quote/uniform?slow=true&part=3"
# pause for 500ms then return
{
    "quote": "'In memory. Very well, then. We, the Party,
              control all records, and we control all memories.
              Then we control the past, do we not?'",
    "source": {
        "chapter": 2,
        "part": 3
    }
}

最基本 asyncio 架構：一個 coroutine

@asyncio.coroutine
def quote_simple(url='/quote/uniform'):
    r = yield from aiohttp.request('GET', url)
    quote_json = yield from r.json()
    return quote_json['quote']

loop = asyncio.get_event_loop()
quote = loop.run_until_complete(quote_simple())

本來使用 requests 長這樣

def quote_block(url='/quote/uniform'):
    r = requests.request('GET', url)
    quote_json = r.json()
    return quote_json['quote']


quote = quote_block()

最基本 asyncio 架構：一個 coroutine

@asyncio.coroutine
def quote_simple(url='/quote/uniform'):
    r = yield from aiohttp.request('GET', url)
    quote_json = yield from r.json()
    return quote_json['quote']

loop = asyncio.get_event_loop()
quote = loop.run_until_complete(quote_simple())

最基本 asyncio 架構：一個 coroutine

@asyncio.coroutine
def quote_simple(url='/quote/uniform'):
    r = yield from aiohttp.request('GET', url)
    quote_json = yield from r.json()
    return quote_json['quote']

loop = asyncio.get_event_loop()
quote = loop.run_until_complete(quote_simple())

Coroutine / Task

• quote_simple() 會回傳一個 coroutine（可以 yield from 它）
• loop.run_until_complete(coro) 會自動把 coroutine 用 Task (subclass of future) 包住， 並丟到 event loop 去執行
• Task 有很多好處：接 callback、看包的 coroutine 有沒有結束、記住回傳值、存 traceback
• 例：t = Task(coro); loop.run_until_complete(t)
  能有相同的作用
• 好處是 t.result() 能拿到 coro 回傳值

多個 requests 情況

用 asyncio.gather(*coro) 同時放入多個 coroutines 進 loop，
等他們結束，並依序收集回傳值。

@asyncio.coroutine
def quote_many(num_quotes=1):
    coroutines = [
        quote_simple() for i in range(num_quotes)
    ]
    quotes = yield from asyncio.gather(*coroutines)
    return quotes

一發不可拾

• loop.run_until_complete(quote_many(2000))
  表示一口氣發了 2000 個 request，會把人家打壞掉……
• 如果電腦沒調整，在那之前會先出現 ClientOSError
• Port 會先不夠用 netstat -apn | grep 9527

tcp 0 0   127.0.0.1:5566    127.0.0.1:45236  ESTABLISHED 5991/python
tcp 0 0   127.0.0.1:45213   127.0.0.1:5566   ESTABLISHED 6680/python
tcp 0 0   127.0.0.1:45168   127.0.0.1:5566   TIME_WAIT   -
tcp 0 0   127.0.0.1:44763   127.0.0.1:5566   TIME_WAIT   -
tcp 0 0   127.0.0.1:45772   127.0.0.1:5566   TIME_WAIT   -
tcp 0 0   127.0.0.1:44741   127.0.0.1:5566   TIME_WAIT   -

Cont'd

• 請參考詳細說明和相關文章調整 sysctl.conf 連線設定
• 實際上跑得很慢，2000 reqs (1 req = 0.5 sec) 需要 4.66 sec
• 很明顯地這沒有解決問題
• 就跟 multithreading 一樣，要加裝拘束器連線限制
• aiohttp 其實可以設定 connection pool limit，但這莫名一直炸 / 噴 Exception。重點最後還是很慢，需要 3.94 sec
• 自己寫。使用 asyncio.Semaphore （維基說明）

本來的 quote_simple

@asyncio.coroutine
def quote_simple(url='/quote/uniform'):
    r = yield from aiohttp.request('GET', url)
    quote_json = yield from r.json()
    return quote_json['quote']

Coroutine with Semaphore

@asyncio.coroutine
def quote_with_lock(semaphore, url='/quote/uniform'):
    with (yield from semaphore):
        r = yield from aiohttp.request('GET', url)
        quote_json = yield from r.json()
    return quote_json['quote']

with ... 進入時 acquire，出來時 release 來控制總活動連線數

@asyncio.coroutine
def quote_many(num_quotes=1, conn_limit=20):
    semaphore = asyncio.Semaphore(conn_limit)
    coroutines = [
        quote_with_lock(semaphore)
        for i in range(num_quotes)
    ]
    quotes = yield from asyncio.gather(*coroutines)
    return quotes

quotes = loop.run_until_complete(
      quote_many(2000, conn_limit=100))

這樣就完成了

• 這樣需要 1.91 sec
• 感覺搞了很久，其實最後的 code 蠻簡單的
• 講到這邊差不多了
• 剩下的就只是爽度問題

進度條真的是好處多多

• 感覺很炫
• 方便知道自己還可以耍廢多久
• 重點是讓自己覺得有進度啊
• 這對士氣有絕對影響啊。誰想盯著一整天都沒有動靜的螢幕看，
  而且一天快要結束的時候，發現自己在收數據的地方出了個 bug，超幹
• IPython Notebook 有 IPython.html.widget.IntProgress；
  console 底下可以拿 pip (6.0+) 裡面的 progressbar 來用

不考慮原本執行順序的話

可以考慮用 asyncio.as_completed(coroutines)。
它會把執行結束的 coroutine 依完成先後丟回對應的 Future

quotes, step = [], 10
for i, fut in enumerate(asyncio.as_completed(coros), 1):
    if i % step == 0:
        progress.value += 1  # or progress.next()
    q = yield from fut  # get the result of future
    quotes.append(q)

考慮執行順序的話：方法一之
包成 Task 呼叫 .result()

futures = [
    asyncio.ensure_future(quote_with_lock(semaphore))
    for i in range(num_quotes)
]
for i, fut in enumerate(asyncio.as_completed(futures), 1):
    if i % step == 0:
        progress.value += 1
    yield from fut  # don't store
quotes = [fut.result() for fut in futures]

方法二之 callback

done_tasks = 0
def progress_adder(fut):
    nonlocal done_tasks
    done_tasks += 1
    if done_tasks % step == 0:
        progress.value += 1

自己做一個 callback function progress_adder 負責增加 progress

（續）包成 Task 加上 callback

futures = []  # wrap coroutines as Tasks
for i in range(num_quotes):
    task = asyncio.Task(quote_with_lock(semaphore))
    task.add_done_callback(progress_adder)
    futures.append(task)

quotes = yield from (asyncio.gather(*futures))

@asyncio.coroutine
def quote_many(num_quotes, conn_limit, progress, step):
    semaphore = asyncio.Semaphore(conn_limit)
    done_tasks = 0
    def progress_adder(fut):
        nonlocal done_tasks
        done_tasks += 1
        if done_tasks % step == 0:
            progress.value += 1

    futures = []
    for i in range(num_quotes):
        task = asyncio.Task(quote_with_lock(semaphore))
        task.add_done_callback(progress_adder)
        futures.append(task)

    quotes = yield from (asyncio.gather(*futures))
    return quotes

Demo: Console Progressbar

Or view it on gfycat

pip 6.0+ 裡面包了各種各樣的 progress bar (Ex. spinner, counter, and bar)

Demo: IPython Notebook Progressbar

Or view it on gfycat

step 設太小，I/O 高且會拖慢速度（最快不超過 0.1s 更新一次）

Take home message

• Aysncio, event loop, and coroutine are not that unfriendly
• ensure_future(coroutine) becomes Task, subclass of Future, which gets many goodies like add_done_callback and .result()
• Multiple concurrent coroutines can be by gather(*coros) or as_completed(coros) to work simultaneously
• Semaphore controls the maximun of active concurrent coroutines

Who the X use Python 3?

• Me. And everyone else 5 years later
• Yes, all previous code requires Python 3.3+
• You mentioned Python 3.5?
  async and await usage at 3_console_demo_py35.py
• You mentioned Python 2.7?
  Tornado has its own event loop and can replace aiohttp.
  Alternatively, Trollius is the asyncio on Python 2. Port is easy.

The REST server by Tornado

class RootHandler(tornado.web.RequestHandler):
    def get(self):
        response = {'message': 'Big Brother is watching you'}
        self.write(response)
        self.finish()

if __name__ == '__main__':
    app = tornado.web.Application([(r'^/$', RootHandler)])
    app.listen(5566)  # try curl -s -XGET localhost:5566/
      

Add extra async operations

By default, Tornado app already is async.
To pause for an extra 0.5 sec before response,

class SlowHandler(RootHandler):
    @tornado.web.asynchronous
    @gen.engine
    def get(self):
        yield gen.Task(IOLoop.current().add_timeout, time.time() + 0.5)
        self.finish()  # close the connection