為什麼需要 async programming 以及相關技術的演進

幾種使用 async programming 的情境：

GUI: 像是Web、iOS、Android apps。在 event-driven 的架構下用 callback 滿直覺的, 但是為了避免卡住 UI thread 而需要連續多個 non-blocking I/O 才能完成一件事時，寫起來就不直覺了。這時新增 thread 用 blocking I/O 寫起來比較直覺，但是不容易處理 thread-safety。

server: 像是 Web server、API server、reverse proxy。要能同時承受上千筆連線，必須用 non-blocking I/O。遇到 CPU 吃重的工作時，可以轉交給其它專門的 server (如向資料庫讀資料或用 Lucene 處理 text search)。

其它像是一些會用到多筆 http request 取資料的小程式，例如用 Gmail API 取出所有信件的附件，或是 stress test 的工具，或是 Web crawler，這些也會用 non-blocking I/O 加速程式 (或用 multi-thread/multi-process + blocking I/O 也行，最近才寫了個小工具 prunner.py)。

共通需求

• 用 non-blocking I/O 避免卡住服務或縮短完成時間。
• 用 single thread 降低實作複雜度。

為了滿足上述需求，主流的解法是用 single-thread event loop。但遇到 CPU bound 時還是得開新 thread。

注意這個共通需求對 Web page 和 server 是必要需求，但對其它情境只是「優先希望的作法」，Android、iOS 或 crawler 還是可以選擇新增 thread/process 使用 blocking I/O，然後自行承擔 multi-thread/multi-process 衍生的問題。

non-blocking I/O 的問題

single-thread event loop 只有幫忙監聽所有 non-blocking I/O 的動靜，用 callback 的方式通知結果。用 callback 處理 GUI 反應很自然，但是需要依序發數個 http request 的時候，在 callback 裡跳來跳去就頭大了，很難看出程式的流程。

換句話說，我們需要以下功能：

• 用 synchronous API 來使用 non-blocking I/O。方便看出控制流程。
• 執行中可以得到 call stack，方便除錯。

提供底層使用 non-blocking I/O 的 synchronous API

如何在不增加新的語法下提供這樣的功能？換句話說，我們還是想新增 thread 用「看起來像 blocking I/O 但底層是 non-blocking I/O」的 API。Python 的 gevent 提供一個不錯的解法：

• 以 event loop 為底，實作 coroutine (由 greenlet 完成)
• 提供 synchronous API + non-blocking I/O 實作 (用 yield 主動讓出執行權)
• task scheduler 會在 I/O 完成後再回到當初 yield 處，讓使用 API 的人用起來像 blocking I/O 一般

剩下的唯一的問題是：所有程式都要用 gevent 提供的 I/O API。實務上這很難作到，所以 gevent 另外提供 monkey patch 處理沒有用 gevent API 的 third-party library。去除 monkey patch 的不穩因素 (比方說 lxml.etree.parse(URL) 會用 native code 從網路讀資料而 block main thread)，效果意外地不錯。除了 Python 以外，Go 也有 goroutine，不過我沒研究也沒相關使用經驗。

值得一提的是，coroutine 是 non-preemptive multitasking，比使用 native thread 的 multi-thread (preemptive multitasking) 容易避免 race condition，不過代價是要小心不要寫出 busy loop 卡住整個 process。

新語法：async/await

相對於 Python 陣營有龐大使用 blocking I/O third-party library 的問題，JavaScript 陣營 (Node.js / Browser) 天生就使用 non-blocking I/O (長年以來只有一個 thread 的自然結果)，這點很有優勢。

JavaScript 不用 coroutine 而是發展出兩種方式提供 synchronous API:

• Promise: 看起來像 synchronous API，只是囉嗦了點，還有可能會搞混執行順序。
• async/await: 基於 Promise 定義新的語法，簡化使用並突顯使用 async 的程式。

詳情見The long road to Async/Await in JavaScript。另外 What Is Asynchronous Programming? 和 Managing Asynchronous Code - Callbacks, Promises & Async/Await 也值得一看。

另外，Python 也會在 3.5 加入新語法 async/await，用它們明確地定義 coroutine，藉此區分 generator。async/await 去掉 thread 的概念，應該會比 coroutine 適合。但對習慣 multi-thread 的人來說，可能 coroutine 比較親切吧？

整體來說，未來繼續用 Node.js 或 Python 寫 server 程式還是不錯的選擇，語法簡單、用戶眾多又有廣大的 third-party library 可用 (目前各用 gevent 和 Node.js 寫過一次上線的服務)。

雜談

話說大家都用 synchronous API 提供 non-blocking I/O 後，還能稱這為 async programming 嗎？不過語法裡有 async/await啦。具體來說，async programming 的定義到底是什麼......

使用 Node.js 實作 server push

前幾天才說用 Tornado 實作 server push 很簡單, 且可以直接用在上線環境裡。今天要來自打嘴巴說我要改用 Node.js 了。有興趣了解相關基本知識的人, 可以參閱之前的說明。

Python 圈 event-based framework 的問題

如這篇文章所述, Python 天生不是 async, 既有函式庫自然也不是 async, 使用 event-based framework (*1) 後, 要另外找配套的函式庫。比方說 http client 用慣了 requests, 直接拿到 Tornado 裡用會出事, 因為 Tornado 是 single process single thread 的架構, 一發出網路連線就會 block 住全部動作。要嘛改用 Tornado 提供的 httpclient, 不然就要找別人用 Tornado 提供的 API 包好的 requests。

每套函式庫都這樣找頗麻煩的 (資料庫、第三方服務的 SDK、etc), 而且不見得找得到牢靠又有效率的版本, 比方說 Google API Python Client 就沒有 Tornado 版的, 要自己想辦法弄。以前用 gevent 時作過類似的事, 先搞定 Dropbox Python SDK, 再來搞不定 Google Drive API, 只好用 gevent monkey patch。感覺不是很可靠。

Node.js 的優勢

反觀 Node.js 天生就是 async, 既有函式庫自然也是 async。Google 也有提供 Google API Node.js Client。既有的社群也滿龐大的, 可以用的套件很多。對我來說, 這點最重要。

效能方面, 因為使用 V8 轉成 native code 執行, 表現也不錯。從這裡和這裡的 benchmark 來看, 數值計算大勝 CPython, 然後和 PyPy 持平。我自己比較常用 dictionary, 作個簡單的 benchmark 測試使用 for, if 和 dictionary, 結果 Node.js 小勝 PyPy, 然後大勝 CPython。整體來說執行速度夠用了, 到是記憶體有 1.7G 的限制 (64bit OS 預設 1G), 要多留意一下。

使用 Express 作 web server 提供簡單的 API, 然後用 ab 作簡單的 benchmark, 結果效能比 Tornado 好 (或可視作差不多, 都達到標準), 一杪內湧入 1k 連線可在一秒內解決; Node.js 底層使用 libuv, 承受 C10k 以上的連線也不是問題。

此外, 配合既有寫網頁的經驗, 改用 Node.js 開發少了學習新語言的成本, 這也是一個優勢。

Server Push 實作

這裡有完整的程式碼, 我只有作最基本功能, 沒控管記憶體也沒提供 timeout 之類的額外參數。寫起來很直覺, /get 沒有資料的時候, 就產生一個 callback function 存到全域變數 g_cache 裡。待 /set 再逐一呼叫這些 callback function, 完成 long polling 的要求。

var error = {
  ok: 0,
  invalid_input: 1,
};

var g_cache = {};

app.get('/set', function (req, res) {
  var key = req.query.key;
  var value = req.query.value;
  if (key === undefined || value === undefined) {
    res.send(JSON.stringify({
      error_code: error.invalid_input
    }));
    return;
  }

  if (!(key in g_cache)) {
    g_cache[key] = {};
  }

  var entry = g_cache[key];
  entry.value = value;
  if ('callbacks' in entry) {
    var cbs = entry.callbacks;
    for (var i = 0; i < cbs.length; i++) {
      cbs[i]();
    }
    while (cbs.length > 0) {
      cbs.pop();
    }
  }
  res.send(JSON.stringify({
    error_code: error.ok
  }));
});

app.get('/get', function (req, res) {
  var key = req.query.key;
  if (key === undefined) {
    res.send(JSON.stringify({
      error_code: error.invalid_input
    }));
    return;
  }

  if (!(key in g_cache)) {
    g_cache[key] = {};
  }

  var entry = g_cache[key];
  if ('value' in entry) {
    res.send(JSON.stringify({
      error_code: error.ok,
      value: entry.value
    }));
  } else {
    if (!('callbacks' in entry)) {
      entry.callbacks = [];
    }
    entry.callbacks.push(function() {
      res.send(JSON.stringify({
        error_code: error.ok,
        value: entry.value
      }));
    });
  }
});

Node.js 雜項備忘

• How to Install Node.js on Ubuntu 14.04
• Installing Node.js via package manager: 這個作法可以裝最新的 stable 版。目前最新的 stable 版本是 0.12, 上個 stable 版本是 0.10, 若要改用 0.10, 不要用 https://deb.nodesource.com/setup_0.12 , 改用 https://deb.nodesource.com/setup_0.10 。我可以用這個作法在 Ubuntu 12.04 上裝 0.10 版。
• How to create an HTTPS server in Node.js?
• Why should I use Express when developing a web app with Node.js?: Express 提供 web server 需要的基本功能如 session, POST, template, error handling, 防止 server crash 等。
• 這裡提到 Express 的架構是: http > connect > Express。有閒時再來看看底層的東西。
• javascript - How to get POST a query in Express.js/Node.js?: Express 4.x 作法不太一樣。摘錄我試成功的方法:

  var express = require('express');
  var app = express();
  
  var bodyParser = require('body-parser')
  app.use(bodyParser.json());       // to support JSON-encoded bodies
  app.use(bodyParser.urlencoded({     // to support URL-encoded bodies
    extended: true
  }));
  
  app.get('/get', function (req, res) {
    res.send(req.query.key);
  });
  
  app.post('/post', function (req, res) {
    res.send(req.body.key);
  });
  
  var server = app.listen(3000, function () {
    var host = server.address().address;
    var port = server.address().port;
    console.log('web server for long polling listening at http://%s:%s', host, port);
  });
  

• Q: A tool for creating and composing asynchronous promises in JavaScrip: 遇到 callback hell 時再來看看。

備註

1. 我指使用 epoll 之類作法的架構。

2015-06-22 更新

Scott 提到 Go 也是不錯的選擇。搜了一下, 還真多人從 Node.js 跑到 Go。另一方面抽查了幾個 third-party 服務, 看起來 Node.js 的支援度比 Go 廣。對我來說, 現階段還是用 Node.js 比較划算 (學習成本和風險考量)。一兩年後再來評估看看是否值得改用 Go。

server push by long polling

程式碼

沒興趣讀落落長心得的人, 這裡有用 Tornado 實作的程式和測試碼。

心得

server push 是指從 server 主動送訊息給 client, 這裡有圖解一些達成 server push 的方式。其中我個人偏好 long polling 的作法，運作方式就和字面一樣: client 先發出一個連線, server 不要立即回應。等 server 需要主動通知 client 時, 再回傳資料。

這個作法的好處是:

• 各種 client 都適用, 只要能用 http 連線即可, 不是瀏覽器也OK, 現在各個平台都有好用的 http 函式庫。
• 不用擔心 client 網路環境問題。client 能主動建立往 server 的連線, 反過來就很難說了。

另一方面, long polling 的缺點是 server 會有許多閒置的連線占資源。像 apache 這類每個 client 用獨立的 thread 處理連線的作法, 就不適合處理 long polling。假設一個 thread 占據 10MB 空間, 1k 個不作事的連線, 就占掉 10G 記憶體了。另外大量 thread 之間的 context switch 也是可觀的時間負擔。

但是改用 epoll 之類的 API 寫成 event-driven 的架構, 可用 single process single thread 的方式同時處理大量連線, 就沒有浪費記憶體和 context switch 的時間負擔。現在有不少現成的工具使用 epoll 包成 green thread, 使用上和 multi-thread 一樣容易上手 (而且還不用擔心 race condition, 因為實際上只有一個 native thread)，降低實作門檻。其中 Tornado 是 Python 寫成的 web framework 並內建可上線用的 web server。我作了簡單的 benchmark, 的確可以快速反應一秒內同時擁入的一千個連線。用 Tornado 省去不少工夫 (若不想用 web server 而是自己從頭作 server, 也可用 gevent)。

實作 long polling 的另一個問題是: 如何在察覺資料更新時, 能立即通知 client? 想像 client 透過連線 S 連往 server, server 先不作回應。接著 server 在其它地方取得新資料, 再透過 S 回傳資料。處理連線 S 的程式要怎麼暫停? 之後要怎麼返回執行? 這個作法可以 scale 嗎?

對於 single process 程式來說, 直覺且低成本的作法是: 在 thread A 讀資料, 發覺沒資料時改用 condition 變數 (搭配 lock) 停住程式, 然後在 thread B 收到新資料時, 用同一個 condition 變數通知 thread A 可以繼續讀資料。對應到 Tornnado 的寫法是用 locks.Condition:

condition = locks.Condition()

@gen.coroutine
def waiter():
    print("I'll wait right here")
    yield condition.wait()  # Yield a Future.
    print("I'm done waiting")

@gen.coroutine
def notifier():
    print("About to notify")
    condition.notify()
    print("Done notifying")

@gen.coroutine
def runner():
    # Yield two Futures; wait for waiter() and notifier() to finish.
    yield [waiter(), notifier()]

io_loop.run_sync(runner)

有了 locks.Condition, 要實作一個完整可上線使用的 long polling 只是小事一椿, 下面的程式碼實作兩個 API /get 和 /set: 提供基本的 "get value by key" 和 "set value for key"。/get 沒資料時會先停住不回應, 等到 /set 取得資料後再立即返回。這只是示意用程式, 沒防錯也沒控管記憶體的用量:

class Entry(object):
   ...

g_cache = {}

class GetHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def post(self):
        global g_cache

        key = self.get_argument("key", None)
        entry = g_cache.get(key, None)
        if not entry:
            entry = g_cache[key] = Entry()

        value = entry.get_value()
        if value is not None:
            self.write(json.dumps({
                'error_code': ERROR_OK,
                'value': value,
            }))
            return

        # Wait the data to be ready.
        timeout = self.get_argument("timeout", DEFAULT_TIMEOUT)
        now = tornado.ioloop.IOLoop.current().time()
        timeout = now + float(timeout)
        condition = entry.get_condition()
        try:
            yield condition.wait(timeout=timeout)  # Yield a Future.
        except Exception, e:
            logging.exception('condition timeout? timeout=%s' % str(timeout))
            self.write(json.dumps({'error_code': ERROR_TIMEOUT}))
            return

        value = entry.get_value()
        self.write(json.dumps({
            'error_code': ERROR_OK,
            'value': value,
        }))


class SetHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def post(self):
        global g_cache

        key = self.get_argument("key", None)
        value = self.get_argument("value", None)
        entry = g_cache.get(key, None)
        if not entry:
            entry = g_cache[key] = Entry()
        entry.set_value(value)
        condition = entry.get_condition()
        condition.notify_all()
        self.write(json.dumps({'error_code': ERROR_OK}))

application = tornado.web.Application([
    (r"/get", GetHandler),
    (r"/set", SetHandler),
])

完整的程式見開頭的連結。

GUI 架構的考量

( 和 Scott 聊過後, 決定改寫《非同步程式新手心得: 為什麼要用單一 thread + task queue 的方式設計架構》的前半段。比較完整地描述 GUI 架構的入門心得。 )

GUI 需要低延遲的反應速度, 隨時能接收使用者輸入和並提供畫面回饋。這表示要在不互相阻檔的前提下應付三種工作:

• 更新畫面
• 偵測使用者輸入
• 處理使用者輸入

甚至第四種情況:

• 網路輸入

若是使用 multi-thread 處理上述情況 (比方說有三個 thread 分別專職處理前述三件工作), 必須留意 thread 之間有共享資料的情況。為了避免 race condition 必須使用 lock; 使用 lock 後又要擔心 dead lock。

使用 global lock 有可能成效不彰, 比方說更新畫面太久, 而沒有即時回饋使用者的輸入。另一方面, 切許多細小的 lock 處理各塊共享資料, 容易出錯造成 dead lock, 或誤以為有處理好卻仍存在 race condition。

反過來說, 若每個工作 (#1) 都切得很細, 執行時間不長, 但可能需要多個工作單元才能完成目標。這樣就能在一個 thread 內以類似 queue consumer 的方式執行全部的工作。有新的需求時, 放入工作到 queue 裡, 大致上可確保相依工作之間的執行順序 (#2)。由於全部工作都在同一 thread 執行, 自然不用擔心 dead lock 或 race condition。

於是可以看到, 像 Gtk、iOS、Android 的平台, 都將更新畫面、偵測事件和處理事件三者整合在 main thread 處理, 並且假設程式以 single thread 的方式執行 (#3)。

我不清楚其它平台的情況, iOS 有提供 NSURLRequest 應付常見的網路需求。讀寫資料和事件的 callback 也都在 main thread 內執行。

這麼一來, 框架開發者省事許多, 使用框架的開發者也可以安心地實作 callback。

備註

#1 可想像每個工作是一個函式, 在 iOS 或 Android 上, 整合這種「工作 queue」的概念到基本框架裡, 用起來很直覺, 希望盡量避免開發者從 thread 的角度去處理事情。

#2 不行的時候只好加入狀態變數了解目前該做什麼, 至少不用擔心狀態變數本身會有 race condition。

#3 若開發者有特別需求, 比方涉及檔案或網路 I/O, 或是需要大量 CPU 計算, 而無法在短時間內完成單一工作的話, 開發者需要自行另外處理。使用新的 thread 的話, 自然需要使用 lock 保護和 main thread 共享的資料, 但需要煩惱的範圍已降致最小。

async 與 non-blocking IO

以前一直沒有分清楚 asynchronous 和 non-blocking 的差異, 寫久了以後才明白兩者是不同層次的東西。

• asynchronous programming: 程式不是一條線走到底, 可能目前做到一個段落停下來, 「過一會兒」再繼續執行
• non-blocking IO: 呼叫讀寫函式時, 不會等到完成才返回

通常要讀寫外部資料時 (一般檔案、網路、pipe 等) , 無法確定返回的時間, 這樣會阻擋程式的運作, 所以希望這個讀取 (或寫入) 不要等完成後才返回 (即「非同步」)。

若不希望此次呼叫有不確定時間的擔擱, 有兩種可能的作法:

• 開新 thread 或用既有的 thread 進行讀寫。總之, 停是停別人家的 thread, 和本 thread 無關
• 設定 file descriptor 為 O_NONBLOCK, 告訴 OS 這個 fd 為 non-blocking IO, 呼叫後要能立即返回

兩者作法的優缺點分析已超出本「隨手記」的範圍 (書本作者愛用的大絕), 需要特別注意的是, 設了 O_NONBLOCK 不表示呼叫 read/write 後就結束了。

man 2 read (或 man 2 write) 的 ERRORS 提到:

• EAGAIN The file descriptor fd refers to a file other than a socket and has been marked nonblocking (O_NONBLOCK), and the read would block.
• EAGAIN or EWOULDBLOCK The file descriptor fd refers to a socket and has been marked nonblocking (O_NONBLOCK), and the read would block. POSIX.1-2001 allows either error to be returned for this case, and does not require these constants to have the same value, so a portable application should check for both possibilities.

意思是說:

• 此招對實體檔案無效 (為了檔案讀寫效能, 一次做完可減少硬碟更動讀寫頭的時間)
• 對 socket 仍有可能失效, 記得用 select / epoll 之類的 system call 請系統在有機會完成 non-blocking 操作時通知你。也就是說, 系統只保證此次呼叫不會 block 你, 不保證此次呼叫一定會成功。由開發者持續嘗試呼叫, 總有完成的時候。

非同步程式新手心得: 為什麼要用單一 thread + task queue 的方式設計架構

最近多了一些 GUI + 非同步IO 的開發經驗, 簡記一下心得。

GUI 架構背後的考量

• 由於 GUI 需要低延遲的反應速度, 必須使用非同步的方式實作。
• Gtk/Android/iOS 都在 main thread 裡處理畫面顯示和事件 callback 是有道理的。藉由架構保證所有 task (繪圖、偵測事件、事件 callback) 在同一 thread 內執行, 寫程式時不用擔心 race condition。反之, 若將繪圖、偵測事件、事件 callback 三者拆到兩個 threads 以上, 會衍生許多潛在問題。
• 共享變數太雜時, 使用 global lock + multi-thread 無法提高太多效能, 失去 multi-thread 的意義。
• 若依各別變數使用不同的 lock 容易寫出問題。
• 在同一個 thread 內執行內全部的 task, 可以避免使用 lock, 相對省事許多。然後用戶有效能需求時, 針對有需求的部份另開 thread 加速, 減少需要使用 lock 的地方。

實作注意事項

• 由於程式不是一路通到底, 不適合用以往的直線方式思考。
• 盡量從環境狀態的角度了解如何處理當下的 task, 而減少保證 task 之間的執行順序。這樣寫起來比較簡單, 也會比較穩定。
• 程式本身或註解要能表示清楚狀態變化對各 task 的影響。提供 debug flag 可以 log 各 task 執行順序, 或註解提及主要的和特別的 task flow, 應該有助於日後維護。
• 若能留下產生每個 task 的 task 為何, 有助於日後在 debugger 內除錯。
• 開始實作以前, 要想清楚可能的狀態變化, 以及留好協助維護的 debug code, 由 debug flag 可以開關。