#eventlet

This recent post, by Adam Young, about using rpdb to debug python CGI processes made me aware of how to enable Keystone to run under HTTPD instead of under an eventlet process.

CPython には GIL があるため、プログラムをマルチスレッドにしてもシングルコア (スレッド) 分の計算リソースしか使うことができない。 マルチコア CPU の性能を生かすには、プログラムをマルチプロセスにする必要がある。 今回は、最近興味をもっている Eventlet の場合 WSGI サーバをマルチプロセスで動作させるにはどうすれば良いのかについて調べてみた。 OpenStack Nova のコードを読んでみたところ、単純にサーバソケットを子プロセスと共有するだけで良さそうだ。 ひとまず Eventlet をインストールする。

$ pip install eventlet

早速、以下サンプル。

#!/usr/bin/env python # -*- coding: utf-8 -*- import multiprocessing import eventlet from eventlet import wsgi def hello_handler(env, start_response): start_response('200 OK', [('Content-Type', 'text/plain')]) return ['Hello, World!\r\n'] def wsgi_serve(server_socket, handler): wsgi.server(server_socket, handler) if __name__ == '__main__': server_socket = eventlet.listen(('localhost', 8080)) for _ in range(multiprocessing.cpu_count() - 1): process = multiprocessing.Process(target=wsgi_serve, args=(server_socket, hello_handler)) process.daemon = True process.start() wsgi_serve(server_socket, hello_handler)

起動してみる。

(30364) wsgi starting up on http://127.0.0.1:8080/ (30366) wsgi starting up on http://127.0.0.1:8080/ (30266) wsgi starting up on http://127.0.0.1:8080/ (30368) wsgi starting up on http://127.0.0.1:8080/

上記は 2 コア (4 スレッド) の MacBook Air で動かしているので 4 つのプロセスが立ち上がった。 次に curl で動作確認してみる。

$ for i in `seq 1 1 10`; do curl http://localhost:8080; done; Hello, World! ...(snip)... Hello, World!

上手く出力されているようだ。 WSGI サーバを起動しているコンソールを見てみる。

(30368) accepted ('127.0.0.1', 51187) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000332 (30266) accepted ('127.0.0.1', 51189) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000319 (30364) accepted ('127.0.0.1', 51191) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000219 (30366) accepted ('127.0.0.1', 51193) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000224 (30364) accepted ('127.0.0.1', 51195) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000097 (30364) accepted ('127.0.0.1', 51197) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000116 (30368) accepted ('127.0.0.1', 51199) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000121 (30364) accepted ('127.0.0.1', 51201) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000086 (30368) accepted ('127.0.0.1', 51203) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000093 (30364) accepted ('127.0.0.1', 51205) 127.0.0.1 - - [28/Oct/2013 22:20:56] "GET / HTTP/1.1" 200 117 0.000090

Eventlet のように、グリーンスレッドで並行処理を行っている場合、そのグリーンスレッド群が動作しているネイティブスレッドがブロックすると処理全体が止まってしまう。 通常は、モンキーパッチによってネイティブスレッドがブロックするコードは全て書き換えられ、ブロックせずに別のグリーンスレッドにコンテキストスイッチするような動作になる。 しかし、これにも一部例外がある。Python は高速化や元々 C で書かれているライブラリの利用などを意図して C 拡張モジュールを使えるようになっている。 C で書かれたモジュールにはモンキーパッチが当てられない。もし C 拡張モジュールの中がブロックするようなコードになっていた場合、これは重要な問題となる。 身近な例では RDB に接続するドライバはよく C 拡張モジュールを使って書かれており、処理の性質上ブロックすることも避けがたい。 では、この問題には解決方法がないのかというと、そんなことはないよというのが今回の内容。 とりあえず Eventlet をインストールしておく。

$ pip install eventlet

まず、次のようなコードを書いてみる。 このコードでは、Eventlet がモンキーパッチを当てる前のオリジナルな time モジュールを使って sleep をかけることで、ネイティブスレッドを意図的にブロックしている。 C 拡張モジュールを書くのが面倒なので、以降はネイティブスレッドをブロックするにはこのやり方を使う。

#!/usr/bin/env python # -*- coding: utf-8 -*- import time import eventlet eventlet.monkey_patch() # モンキーパッチを当てる前のオリジナルのモジュールを取り出す from eventlet import patcher orig_time = patcher.original('time') def greenthread1(): while True: print('Green thread: 1') time.sleep(1) def greenthread2(): while True: print('Green thread: 2') time.sleep(1) if __name__ == '__main__': eventlet.spawn(greenthread1) eventlet.spawn(greenthread2) while True: print('Native thread: 1') # グリーンスレッドが動いているネイティブスレッドをブロックさせる orig_time.sleep(1)

これを実行すると以下のようになる。

Native thread: 1 Native thread: 1 Native thread: 1

ネイティブスレッドがブロックしている上に、グリーンスレッドは切り替えのタイミングも与えられていないため、全く並行処理が行われなくなってしまっている。 一応、上記のコードでも、グリーンスレッドを切り替えるタイミングさえ与えてやれば、並行処理が全くできないわけではない。 例えば、上記を少しだけ変更して、以下のようにしてみる。

#!/usr/bin/env python # -*- coding: utf-8 -*- import time import eventlet eventlet.monkey_patch() # モンキーパッチを当てる前のオリジナルのモジュールを取り出す from eventlet import patcher orig_time = patcher.original('time') def greenthread1(): while True: print('Green thread: 1') time.sleep(1) def greenthread2(): while True: print('Green thread: 2') time.sleep(1) if __name__ == '__main__': eventlet.spawn(greenthread1) eventlet.spawn(greenthread2) while True: print('Native thread: 1') # グリーンスレッドが動いているネイティブスレッドをブロックさせる orig_time.sleep(3) # 明示的にグリーンスレッドを切り替える time.sleep(0)

上記を実行する。

Native thread: 1 Green thread: 1 Green thread: 2 Native thread: 1 Green thread: 1 Green thread: 2 Native thread: 1

今度はグリーンスレッドの並行処理も一応は進んでいる。 しかし、本来意図している 1 秒毎の処理ではなく 3 秒毎に表示されているようだ。 これは、ネイティブスレッド自体がブロックしている 3 秒の間はグリーンスレッドは処理が行えていないため。 つまり、グリーンスレッドは自身が動作しているネイティブスレッドがブロックしてしまうと、いずれにせよ都合が悪い。 なので、グリーンスレッドとネイティブスレッドを併用する場合には、グリーンスレッドが動作しているのとは別のネイティブスレッドを走らせる。 次のコードでは、ネイティブスレッドをブロックする処理の実行を、グリーンスレッドが動作しているのとは別のネイティブスレッド上で実行させてみる。

#!/usr/bin/env python # -*- coding: utf-8 -*- import time import eventlet eventlet.monkey_patch() # モンキーパッチを当てる前のオリジナルのモジュールを取り出す from eventlet import patcher orig_threading = patcher.original('threading') orig_time = patcher.original('time') def greenthread1(): while True: print('Green thread: 1') time.sleep(1) def greenthread2(): while True: print('Green thread: 2') time.sleep(1) def nativethread1(): while True: print('Native thread: 1') orig_time.sleep(3) def native_spawn(): native_thread = orig_threading.Thread(target=nativethread1) native_thread.start() if __name__ == '__main__': eventlet.spawn(greenthread1) eventlet.spawn(greenthread2) native_spawn() while True: time.sleep(1)

上記を実行する。

Native thread: 1Green thread: 1 Green thread: 2 Green thread: 1 Green thread: 2 Green thread: 1 Green thread: 2 Native thread: 1 Green thread: 1 Green thread: 2 Green thread: 1 Green thread: 2 Green thread: 1 Green thread: 2 Native thread: 1

コンソールが (ネイティブ) スレッドセーフではないため、一部表示が崩れているが、なんにせよ本来意図していたようにネイティブスレッドは 3 秒間隔、グリーンスレッドは 1 秒間隔で表示を続けている。 ネイティブスレッドがブロックする処理はグリーンスレッドが動作しているのとは別のネイティブスレッドで実行すれば良いことがわかった。 次はグリーンスレッドとネイティブスレッドの間でネイティブスレッドをブロックするようなジョブの受け渡しをしてみる。

#!/usr/bin/env python # -*- coding: utf-8 -*- import eventlet eventlet.monkey_patch() import time import random from eventlet.event import Event # モンキーパッチを当てる前のオリジナルのモジュールを取り出す from eventlet import patcher orig_threading = patcher.original('threading') orig_time = patcher.original('time') orig_Queue = patcher.original('Queue') # グリーンスレッド・ネイティブスレッド間で仕事の要求・結果を受け渡しするためのキュー _REQUEST_Q = orig_Queue.Queue(maxsize=-1) _RESPONSE_Q = orig_Queue.Queue(maxsize=-1) # ジョブをネイティブスレッドに依頼するグリーンスレッド def put_green_thread(): while True: e = Event() # 1 から 3 秒の間でランダムにネイティブスレッドをブロックするジョブを作る job = (e, blocking_native_thread_job, random.randint(1, 3)) # ジョブを登録する _REQUEST_Q.put(job) # ジョブの完了を待つ result = e.wait() # 結果を表示する print(result) # ネイティブスレッドで処理が完了したジョブを取り出すグリーンスレッド def get_green_thread(): while True: while not _RESPONSE_Q.empty(): try: # 処理済みのジョブはノンブロックで取り出す (重要) e, result = _RESPONSE_Q.get(block=False) except orig_Queue.Empty: pass # ジョブが完了したことを待っているグリーンスレッドに通知する e.send(result) # グリーンスレッドを切り替える time.sleep(0) # C 拡張モジュール内で行われるような、ネイティブスレッドをブロックする仕事 def blocking_native_thread_job(block_time): # 一定時間ネイティブスレッドをブロックする orig_time.sleep(block_time) # 処理結果としてメッセージを返す return 'Done, blocked %d sec' % block_time # ジョブを処理するネイティブスレッド def native_thread(): while True: # ジョブをキューから取り出す e, job, args = _REQUEST_Q.get(block=True) # ジョブをネイティブスレッドで処理する result = job(args) # 処理結果をキューに詰める _RESPONSE_Q.put((e, result)) def native_spawn(): thread = orig_threading.Thread(target=native_thread) thread.start() def print_green_thread(): while True: print('Green thread') time.sleep(1) if __name__ == '__main__': eventlet.spawn(put_green_thread) eventlet.spawn(get_green_thread) eventlet.spawn(print_green_thread) native_spawn() while True: time.sleep(1)

実行してみる。

Green thread Green thread Done, blocked 1 sec Green thread Green thread Green thread Done, blocked 3 sec Green thread Done, blocked 1 sec

上手く動いてそう。 ただ、上記のコードは正直言ってかなり面倒くさい。 実は同様のことを実現する機能は既に Eventlet で用意されている。 eventlet.tpool を使うことでネイティブスレッドのスレッドプールを経由して処理を実行できる。

#!/usr/bin/env python # -*- coding: utf-8 -*- import eventlet from eventlet import tpool eventlet.monkey_patch() # モンキーパッチを当てる前のオリジナルのモジュールを取り出す from eventlet import patcher orig_time = patcher.original('time') import time import random # ジョブをネイティブスレッド (プール) に依頼するグリーンスレッド def put_green_thread(): while True: result = tpool.execute(blocking_native_thread_job, random.randint(1, 3)) # 結果を表示する print(result) def print_green_thread(): while True: print('Green thread') time.sleep(1) # C 拡張モジュール内で行われるような、ネイティブスレッドをブロックする仕事 def blocking_native_thread_job(block_time): # 一定時間ネイティブスレッドをブロックする orig_time.sleep(block_time) # 処理結果としてメッセージを返す return 'Done, blocked %d sec' % block_time if __name__ == '__main__': eventlet.spawn(put_green_thread) eventlet.spawn(print_green_thread) while True: time.sleep(1)

前述したコードよりも、自分で書いていない分、だいぶスッキリした。 実行してみる。

Green thread Green thread Green thread Done, blocked 3 sec Green thread Green thread Done, blocked 1 sec Green thread

ネットワーキングライブラリを標榜して、主にネットワークを使ったアプリケーションで使われる Eventlet だけど、最近は別にネットワークを使わないアプリケーションで処理の並列化のために使うのも便利なんじゃないか？と思えてきた。 ひとまず Eventlet のインストールは PyPI から。

$ pip install eventlet

以下は全くネットワークは使っていないけど Eventlet を使って処理を並列化しているサンプルコード。

#!/usr/bin/env python # -*- coding: utf-8 -*- import eventlet eventlet.monkey_patch() import time def greenthread1(): while True: print('Hello') time.sleep(1) def greenthread2(): while True: print('World') time.sleep(1) if __name__ == '__main__': eventlet.spawn(greenthread1) eventlet.spawn(greenthread2) while True: time.sleep(1)

実行すると Hello と World が 1 秒間隔で交互に出力される。

Hello World Hello World …

これは何故動作するかというと、Eventlet が既存のブロックを引き起こすコードにモンキーパッチを当てる (eventlet.monkey_patch() の部分) ことで、それらを非同期 I/O に対応させているため。 動作原理から説明すると、Eventlet はユーザーランドで実装されたスレッド (グリーンスレッド) 単位で動いているために、カーネルランドで実装されたスレッド (ネイティブスレッド) とは違って、コンテキストスイッチ (スレッドの切り替え) は自動では発生しない。 じゃあ、一体いつグリーンスレッドの切り替えが行われるかというと、本来はブロックするコードに到達した時点に限られる。 上記のコードを見ると、行の出力を終えたタイミングで time.sleep(1) を行い (本来の意図では) 1 秒間 (ネイティブ) スレッドを停止させている。 (ネイティブ) スレッドが停止する、ということはブロックが発生しているということだ。 本来であれば time.sleep() はネイティブスレッドがブロックして以上オシマイなわけだけど、このコードに Eventlet のモンキーパッチが当たることで挙動を変更している。 つまり、time.sleep() を発行した内部では、一時的にユーザーランドのスタックを別の領域に退避させ、現在実行可能な他のグリーンスレッドを探し、あればそのスタックを復元させて実行する、といった挙動に置き換わっている。 より分かりやすいように以下のようなコードを書いてみた。

#!/usr/bin/env python # -*- coding: utf-8 -*- import eventlet eventlet.monkey_patch() import time def greenthread1(): print('g1-1') print('g1-2') time.sleep(0) print('g1-3') print('g1-4') def greenthread2(): print('g2-1') print('g2-2') time.sleep(0) print('g2-3') print('g2-4') if __name__ == '__main__': eventlet.spawn(greenthread1) eventlet.spawn(greenthread2) while True: time.sleep(1)

実行すると以下のような結果になる。

g1-1 g1-2 g2-1 g2-2 g1-3 g1-4 g2-3 g2-4

最近、使い勝手の良さからぼくの中で評価が赤丸急上昇中のネットワーキングライブラリ Eventlet には Timeout という便利なコンテキストマネージャがある。 これはステートメント内の処理が一定時間内に終わらない場合に例外を上げてくれるというもので、スレッド間の同期を取る際のタイムアウト処理などに使える。 Eventlet のインストールは PyPI から。

$ pip install eventlet

以下のサンプルはタイムアウトが 1 秒にも関わらずステートメント内の処理に 5 秒かかる、というパターン。

#!/usr/bin/env python # -*- coding: utf-8 -*- import eventlet eventlet.monkey_patch() import time from eventlet.timeout import Timeout if __name__ == '__main__': try: with Timeout(1): time.sleep(5) except Timeout: print('Over the time limit')

実行すると時間切れを起こして例外が上がることが分かる。

Over the time limit

なるほど、これは便利だなと思い、ステートメント内の処理が一定時間内に終わった場合に、残りの時間を待ち合わせるコンテキストマネージャ FixedTime を試しに作ってみた。

#!/usr/bin/env python # -*- coding: utf-8 -*- import time from contextlib import contextmanager from threading import Timer, Event class DeadlineException(BaseException): pass @contextmanager def FixedTime(seconds): def _dead_line(ev): ev.set() start = time.time() ev = Event() timer = Timer(seconds, _dead_line, args=[ev]) timer.start() yield timer.cancel() if ev.is_set(): raise DeadlineException() processing_time = (time.time() - start) time.sleep(seconds - processing_time) if __name__ == '__main__': # ステートメント内の処理が n 秒未満で終わる場合にも n 秒はかかるようにする s = time.time() with FixedTime(5): time.sleep(1) ptime = (time.time() - s) print('Processing time is %s sec' % ptime) # ステートメント内の処理が n 秒以上かかった場合には例外をスローする try: with FixedTime(1): time.sleep(5) except DeadlineException: print('Over the time limit')

実行してみる。

Processing time is 4.99916100502 sec Over the time limit

Python での開発には主に Eclipse の PyDev プラグインを使っているんだけど、このデバッガは GUI を使ってデバッグできるので大変便利。 ただ、ネットワーキングフレームワークの Eventlet と組み合わせて使うときにはちょっとした注意が必要。 大抵の場合 Eventlet を使うときは、ブロックを起こす既存のコードにモンキーパッチを当てることでノンブロッキング I/O に対応させる。 この過程で、スレッドにもパッチが当たっていると PyDev のデバッガが正常に動作しない。 そこで、以下のようにモンキーパッチを当てる際にスレッドは除外する必要がある。

eventlet.monkey_patch(thread=False)

Eventlet はコルーチンライブラリ Gleenlet とノンブロッキング I/O を組み合わせたネットワーキングライブラリ。 ノンブロッキング I/O にありがちなコールバックスタイルのコードは可読性が低くなってしまいがちだけど Eventlet ではそこにコルーチンを使うことでブロッキング I/O のコードと同様の記述を可能にすることでそれを補っている。 また、ノンブロッキング I/O のスレッドの中では、ブロックするコードを書くと全体が止まってしまう問題があるけど Eventlet では既存のブロックするコードをブロックしないように書き換えるモンキーパッチを当てることで対処している。 インストール インストールは pip を使って PyPI から。

$ pip install eventlet

Echo サーバ クライアントからの入力をそのまま返す Echo サーバを Eventlet を使って実装してみる。

#! /usr/bin/env python # -*- coding: utf-8 -*- import eventlet def echo_handler(sock, addr): # ソケットをファイルライクオブジェクトに変換する fd = sock.makefile() while True: line = fd.readline() if not line: break # 入力をそのまま返す (Echo) fd.write(line) fd.flush() if __name__ == '__main__': server_sock = eventlet.listen(('localhost', 3001)) # クライアントからの accept() 毎にハンドラが起動する eventlet.serve(server_sock, echo_handler)

ソケット関連の処理は eventlet でラップされている。 上記を socat を使って動作確認してみる。socat についてはこちらを参照のこと。

$ socat STDIN tcp-connect:localhost:3001

上記の処理はシングルスレッドだが、ノンブロッキング I/O を使って処理が並列化されているため、複数のクライアント (socat) を立ち上げてもサーバが処理できる。 Chat サーバ 同じ要領で Chat サーバを書いてみる。

#! /usr/bin/env python # -*- coding: utf-8 -*- import eventlet PORT = 3001 fds = [] def chat_handler(sock, addr): print('Participant joined chat.') fd = sock.makefile() fds.append(fd) _loop_handler(fd) fds.remove(fd) print('Participant left chat') def _loop_handler(fd): while True: line = fd.readline() print('Chat: %s' % (line.strip(), )) if not line: break _broadcast_message(fd, line) def _broadcast_message(pfd, msg): for fd in fds: # 自分の発言はエコーバックしない if fd is pfd: continue fd.write(msg) fd.flush() if __name__ == '__main__': print('Server starting up on port %d' % (PORT, )) listener = eventlet.listen(('localhost', PORT)) eventlet.serve(listener, chat_handler)

動作確認は同様に socat で行う。

$ socat STDIN tcp-connect:localhost:3001

Echo サーバと違って複数クライアントを立ち上げると別のクライアントの発言がブロードキャストされる。 WSGI サーバ 次は WSGI サーバのコンテナ部分を Eventlet にしてみる。 アプリケーション部分は Flask で書く。 Flask についてはこっちを参照。

#! /usr/bin/env python # -*- coding: utf-8 -*- from flask import Flask import eventlet from eventlet import wsgi app = Flask(__name__) @app.route('/', methods=['GET']) def index(): return 'Hello, World!!' if __name__ == '__main__': server_sock = eventlet.listen(('localhost', 8080)) wsgi.server(server_sock, app)

動作確認は curl とかで。

$ curl http://localhost:8080/ Hello, World!!