Pythonの非同期プログラミングを完全理解
非同期プログラミング
非同期プログラミングについて、聞いたことのある人は多いと思います。例えば、フロントエンドで使われているJavaScriptはシングルスレッド言語で、メインスレッドをブロッキングさせないため、様々な関数は非同期処理になるよう実装されています。Node.jsもその性質を受け継ぎ、I/Oバウンドタスクに長けています。しかし、Pythonになると、並列・並行処理に対応しているため、ほとんどの人は自身のプロジェクトで非同期プログラミングを利用した経験がないでしょう。もちろん、Tornado、TwistedやGeventなどの非同期フレームワークが有名で使ったことのある人は少なくないですが、変わったエラーにぶつかった時はなかなか解決できないでしょう。
近年のPyConの傾向から見れば分かりますが、非同期プログラミングは間違いなくPythonエコシステムの次のトレンドになります。また、新興のプログラミング言語であるGo、Rustなどは非同期処理と高性能並行処理を売りにしています。Pythonも負けてはならないので、2013年からPythonの生みの親であるGuidoさん自らTulip(asyncio)プロジェクトの開発を行い始めました。
1. 非同期処理とは
まず、関連するコンセプトを説明してから、非同期処理の説明に入りたいと思います。
1-1. ブロッキング
- プログラムは計算リソースを得られずハングアップになった状態
- プログラムはある処理が完成するのを待っている時、自身は他の処理を行えない状態をブロッキングという
- よく見られるブロッキングの形式:ネットワークI/Oブロッキング、ハードディスクI/Oブロッキング、ユーザー入力ブロッキング
ブロッキングはあらゆるところに存在します。例えば、CPUが「コンテキストスイッチ」をしている時、他のプロセスは処理を行えないため、ブロッキングされます。(マルチコアの場合は、コンテキストスイッチをしているコアが利用不可になる)
1-2. ノンブロッキング
- プログラムはある処理を待つ時、自身はブロッキングされず、他の処理に移れる状態をノンブロッキングという
- ノンブロッキングは全てのプログラムレベルと状況で可能なものではない
- プログラムのカプセル化レベルにおいて、子のプログラムユニットを許容できる時(「不可分処理」ではない時)のみ、ノンブロッキング状態が存在しうる
ノンブロッキングはブロッキングの反対面で、ある処理のブロッキングが計算効率を低下させたため、その処理をノンブロッキングにするわけです。
1-3. 同期処理
- 異なるプログラムユニットはあるタスクを完成するため、実行過程において何らかの通信方式で一貫性を保持している時、これらのプログラムユニットを同期処理という
- 例えば、ECサイトのDBで商品在庫の更新において、「行ロック」を信号として、それぞれの更新リクエストに強制的に順序付ける時、この在庫更新の処理を同期処理と言える
- すなわち、同期処理は順番のある処理と言える
1-4. 非同期処理
- 異なるプログラムユニットはお互い通信しなくても、あるタスクを処理できる方式
- 関連性のないプログラムユニット間は非同期である
- 例えば、Webページをスクレイピングする場合、スケジューリングプログラムがダウンロード用のプログラムを呼び出したら、そのダウンロードタスクとの通信を保持する必要なく、すぐ他のタスク用のプログラムを呼び出すことができます。また、異なるWebページのダウンロードなどの処理はお互い関連性がないため、通知し合う必要もありません。ただし、これらの非同期処理は完成時間が未知です。
- すなわち、非同期処理は順番のない処理と言える
上記の通信方式というのは非同期処理と並行処理においての「synchronization primitive」を指します。例えば、セマフォ、ロック、同期キューなどがあります。これらの通信方式は、複数のプログラムをある条件の下で同期処理を行うためのものです。そして、非同期処理があるからこそ、これらの通信方式が必要になるわけです。理由として、全てのプログラムが同期処理を行うのなら、もともと順番に処理されるため、通信方式なんかは必要ないからです。
1-5. 並行処理
- 並行処理はプログラムの構造で、プログラムを複数の独立で実行できる子タスクとして設計することを指す
- 有限の計算リソースを利用し、マルチタスクを同時処理、または近似的に同時処理を行うのが目的
1-6. 並列処理
- 並列処理はプログラムの実行状態で、マルチタスクが同時処理されることを指す
- 余分の計算リソース(マルチコアCPU)を利用し、マルチタスクの実行を加速させるのが目的
並列・並行処理については、前回の記事を参照してください。
1-7. コンセプトのまとめ
- 並列処理はマルチコアCPUを利用して、マルチタスクを加速させるためのもの
- 並行処理は独立した子タスクをできるだけ速く実行するためのもので、全体の処理を加速できるとは限らない
- ノンブロッキングはプログラムの実行効率を向上させるためのもの
- 非同期処理は効率よくノンブロッキングタスクを管理するためのもの
並行処理をサポートするためにプログラムをマルチタスクに分割する必要があります。それぞれのタスクに対して、ブロッキング・ノンブロッキング、同期・非同期が定義されます。そのため、並行、非同期、ノンブロッキングの三者は密接した関係性を持っています。
1-8. 非同期プログラミング
- 「プロセス」、「スレッド」、「コルーチン」、関数・メソッドなどをプログラムの基本ユニットとして、「コールバック」、「イベントループ」、「セマフォ」などのメカニズムを利用し、プログラム全体の実行効率と並行処理性能を向上するプログラミング方式
同期・非同期、ブロッキング・ノンブロッキングは相容れないものではありません。例えば、ECサイトはマルチユーザーのアクセスリクエストに対して非同期処理を行うが、在庫の更新は同期処理である必要があります。
1-9. 非同期プログラミングの難点
- コントロールが効かない。理由は、実行順序は予測不可能だから。並行処理を加えるとさらに難しくなる
そのため、非同期フレームワークのほとんどは非同期プログラミングモデルをシンプル化(一度に1つのイベントしか処理を許容しない)しています。非同期プログラミングについての議論はシングルスレッドのものに集中しています。
- あるイベントを処理するプログラムは長い計算時間を必要とするとき、他の部分は全部ブロッキングされる
そのため、非同期プログラミングを採用したら、各非同期呼び出しを小さくしないとだめです。ここの「小さく」というのは計算時間を短くすることです。そして、プログラムをどのように非同期タスクに分割するのが難題になります。
- プログラムの次のステップは前のステップの結果に依存することが多い。非同期呼び出しの時、その結果または完成しているのかを知る方法が必要
- コールバックは必然になるが、「コールバック地獄 (callbackhell)」に陥って苦痛
- 同期プログラミングのコードを非同期プログラミングに変更すると、コードの構造は必然的にぶっ壊れる
- 問題を解決するためのシンキングも変えなければならない。ただただ突っ走るのではなく、細心の注意を払って非同期タスクを配置する必要がある
2. 非同期プログラミングを採用する理由
前述のように、非同期プログラミングにはたくさんの難点があります。Pythonの生みの親が4年もかけて自ら取り組んで作った「asyncio」はPython 3.6で標準ライブラリーになったが、なんであんな苦労してまで作らなければならないでしょうか?理由は、非同期プログラミングは非常に役に立つものだからです。
2-1. CPUの時間感覚
CPUのクロック数を2.6Ghz
と仮定し,つまり1秒あたり0.38ns
になります。0.38ns
がCPUの時間感覚において最小単位になります。そして、人間の時間感覚の最小単位を1s
とします。下の表は、人間の時間感覚をベースにCPUの時間感覚を計算したものになります。また、CPUの遅延データは「Latency numbers every programmer should know」を参照しています。
処理 | 実際の遅延 | CPUの時間感覚 |
---|---|---|
命令処理 | 0.38ns | 1s |
L1キャッシュの読み込み | 0.5ns | 1.3s |
分岐予測修正 | 5ns | 13s |
L2キャッシュの読み込み | 7ns | 18.2s |
排他制御 | 25ns | 1m5s |
メモリー参照 | 100ns | 4m20s |
コンテキストスイッチ | 1.5µs | 1h5m |
1Gbpsのネットワークで2KBのデータをアップロード | 20µs | 14.4h |
メモリから1MBの連続データを読み込む | 250µs | 7.5day |
同じインターネットデータセンターのホストにpingする(往復) | 0.5ms | 15day |
SSDから1MBの連続データを読み込む | 1ms | 1month |
ハードディスクから1MBの連続データを読み込む | 20ms | 20month |
違う都道府県のホストにpingする(往復) | 150ms | 12.5year |
バーチャルマシンの再起動 | 4s | 300year |
サーバーの再起動 | 5m | 25000year |
CPUはコンピューターの処理コアで、貴重なリソースになります。CPUの実行時間を無駄遣いし、利用率を低下させると、プログラムの効率も必然的に低下します。上記の表が示したように、1Gbpsのネットワークで2KBのデータをアップロードしたら、CPUの感覚では14時間を過ごしたようなものです。もし、10Mbpsのネットワークとなると、更に100倍も効率が下がります。この長い時間を、CPUをただただ待たせて他の処理に移さない行為はまさにCPUの「青春」の無駄遣いになります。
2-2. 現実の問題点
- コスト
プログラムはコンピューターの計算リソースを有効に利用できないと、その穴を埋めるために、より多くのコンピューターが必要になってきます。例えば、スクレイピングのプログラムを非同期プログラミングで設計し直すと、もともと必要とした7台のサーバーを3台まで減らせて、コストを57%削減することができます。ちなみに、AWSではm4.xlargeのリザーブドインスタンスは一年あたり15万円ぐらいかかります。
- 効率
お金はどうでも良いなら、効率はどうしても気になるでしょう。サーバーの数を一定量まで増やしたら、アーキテクチャーやプログラムのデザインを改善しないと、更に増やしても性能が上がらないことがあります。そして、管理コストが圧倒的に増えることになります。
- C10K/C10M問題
「C10K問題」は1999年に提出されたもので、どのようにすれば、1GhzのCPU、2Gメモリと1Gbpsのネットワーク環境で、シングルサーバーで1万のクライアントに同時にFTPサービスを提供することができるだろうというチャレンジみたいなものです。2010年から、ハードウェアの性能の向上により、C10Kに続き、「C10M問題」が提出されました。C10M問題は、8コアのCPU、64Gメモリと10Gbpsのネットワーク環境で、1秒あたり100万の同時アクセスを処理する問題です。
コストと効率は企業経営の角度からの問題で、C10K/C10M問題はハードウェアに対する技術的なチャレンジになります。C10K/C10M問題が解決できたら、コストと効率の問題も同時に解決されることになります。
2-3. 解決方法
CPUは計算が非常に速いが、コンテキストスイッチ、メモリ読み込み、ハードディスク読み込み、ネットワーク通信が非常に遅いです。つまり、CPUから離れると、L1キャッシュ以外は全部遅いです。コンピューターは5大装置の入力装置、出力装置、記憶装置、制御装置、演算装置からなりますが、制御装置と演算装置はCPUの中にありますが、それ以外は全部I/Oです。メモリの読み書き、ハードディスクの読み書き、ネットワークインタフェースカードに対する読み書きは全てI/Oになります。I/Oは最大なボトルネックになっています。
非同期プログラムは効率を上げることができますが、最大なボトルネックはI/Oなので、それに対するソリューションは「非同期I/O」と呼びます。
3. 非同期I/Oの進化の道
この地球上、一番規模の大きいプログラムはインターネットでしょう。そして、CPUの時間感覚の表から、ネットワークI/OはハードディスクI/Oよりも遅く、最大なボトルネックになっていることが分かります。サーバーの再起動などを除き、ネットワークI/Oより遅いものはないため、様々な非同期フレームワークはネットワークI/Oを狙っています。
スクレイピングを例として説明します。ここでは、ネットから10個のWebページをダウンロードします。
3-1. 同期ブロッキング方式
一番簡単なのは順番にダウンロードする方式です。socket
の接続の確立からリクエスト送信、レスポンス受信という順番で実行されます。
import socket
import time
def blocking_way():
sock = socket.socket()
# blocking
sock.connect(('example.com', 80))
request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'
sock.send(request.encode('ascii'))
response = b''
chunk = sock.recv(4096)
while chunk:
response += chunk
# blocking
chunk = sock.recv(4096)
return response
def sync_way():
res = []
for i in range(10):
res.append(blocking_way())
return len(res)
if __name__ == '__main__':
elapsed_times = 0
for _ in range(10):
start = time.time()
sync_way()
elapsed_time = time.time() - start
elapsed_times += elapsed_time
print(f"elapsed_time: {(elapsed_time):.2f}[sec]")
print(f"mean_elapsed_time: {(elapsed_times/10):.2f}[sec]")
実行結果:
elapsed_time: 2.76[sec]
elapsed_time: 2.58[sec]
elapsed_time: 2.56[sec]
elapsed_time: 2.58[sec]
elapsed_time: 2.85[sec]
elapsed_time: 2.66[sec]
elapsed_time: 2.60[sec]
elapsed_time: 3.38[sec]
elapsed_time: 2.88[sec]
elapsed_time: 2.67[sec]
mean_elapsed_time: 2.75[sec]
10回の平均時間は2.75秒です。blocking_way()
関数はsocket
接続を確立し、HTTPリクエストを送り、socket
からHTTPレスポンスを読み取り、データを返します。そして、sync_way()
関数はそれを10回繰り返しただけです。上記のコードでは、sock.connect(('example.com', 80))
でサーバーの80番に対してリクエストを送り、sock.recv(4096)
でsocket
から4KBずつバイトデータを読み取ります。
ネットワークの接続がいつ確立されるのはクライアント側が決めることではなく、ネットワーク環境やサーバーの処理能力によって決まるものです。そして、サーバーからいつデータが返ってくるのも予測できないです。そのため、デフォルトではsock.connect()
とsock.recv()
はブロッキングします。一方で、sock.send()
は長くブロッキングはしません。sock.send()
はリクエストデータをTCP/IPプロトコルスタックのバッファにコピーしたらすぐ戻り値を返すため、サーバーからのレスポンスを待たないです。
もし、ネットワーク環境が非常に悪くて、接続の確立に1秒がかかるとしたら、sock.connect()
は1秒ブロッキングします。この1秒は2.6GHzのCPUにとって、83年のような時間感覚になります。この83年間、CPUは何もできません。sock.recv()
も同じく、サーバーからのレスポンスをクライアントが受信するまで待たなければなりません。example.comのホームページを10回ダウンロードしたら、このブロッキングは10回繰り返します。では、一日1000万のWebページをダウンロードする大規模なスクレイピングはどうなるでしょう。
まとめると、同期ブロッキング方式のようなネットワークI/Oは、効率が非常に低くて、頻繁に通信するプログラムの中ではなおさらです。このような方法は、C10K/C10Mを解決できるわけがありません。
3-2. 同期ブロッキング方式の改良:マルチプロセス方式
同じプログラムで10回実行するのは時間がかかるなら、10個同じプログラムを同時に実行させれば良いでしょう。そこで、マルチプロセスを導入します。ちなみに、Linux 2.4以前のOSにおいては、プロセスこそがタスクのエンティティであり、OSはプロセス指向にデザインされたのです。
import socket
import time
from concurrent.futures import ProcessPoolExecutor
def blocking_way():
sock = socket.socket()
# blocking
sock.connect(('example.com', 80))
request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'
sock.send(request.encode('ascii'))
response = b''
chunk = sock.recv(4096)
while chunk:
response += chunk
# blocking
chunk = sock.recv(4096)
return response
def multi_process_way():
with ProcessPoolExecutor(max_workers=10) as executor:
futures = {executor.submit(blocking_way) for i in range(10)}
return len([future.result() for future in futures])
if __name__ == '__main__':
elapsed_times = 0
for _ in range(10):
start = time.time()
multi_process_way()
elapsed_time = time.time() - start
elapsed_times += elapsed_time
print(f"elapsed_time: {(elapsed_time):.2f}[sec]")
print(f"mean_elapsed_time: {(elapsed_times/10):.2f}[sec]")
実行結果:
elapsed_time: 0.49[sec]
elapsed_time: 0.47[sec]
elapsed_time: 0.48[sec]
elapsed_time: 0.49[sec]
elapsed_time: 0.54[sec]
elapsed_time: 0.51[sec]
elapsed_time: 0.56[sec]
elapsed_time: 0.52[sec]
elapsed_time: 0.47[sec]
elapsed_time: 0.47[sec]
mean_elapsed_time: 0.50[sec]
10回の平均時間は0.50秒です。効果がありましたね。ただし、同期ブロッキング方式の10分の1になってないのが問題です。その理由は実行環境のCPUは10コアではなく、プロセスの切り替えが必要だからです。
プロセスの切り替えはCPUの時間感覚の「コンテキストスイッチ」項目で示したように安くない処理です。CPUは1つのプロセスからもう1つのプロセスに切り替える時、まず、元のプロセスランタイムのレジスタ状態、メモリ状態を全部保存し、それからもう1つのプロセスの保存した状態を復元します。CPUにとっては、数時間待たされたようなものです。プロセス数がCPUのコア数より多い時は、プロセスの切り替えが必ず必要になります。
切り替え以外に、マルチプロセスにはもう1つの欠点があります。普通のサーバーは安定した状態で作動するため、同時処理できるプロセスの数は数十から数百規模に留まっています。プロセス数があまりにも多いと、システムが不安定になり、メモリリソースも足りなくなります。
そして、マルチプロセスは切り替えや規模の小さ以外にも、状態やデータの共有などの問題があります。
3-3. 同期ブロッキング方式の更なる改良:マルチスレッド方式
スレッドのデータ構造はプロセスより軽量で、プロセス内に複数のスレッドを持つことができます。Linux 2.4より新しいOSも、スケジューリングできる最小単位をプロセスからスレッドに変更しました。プロセスはただスレッドの容器として存在し、リソースの管理という役割を果たすようになりました。OSレベルのスレッドはCPUのそれぞれのコアに分配され、同時実行できるものです。
import socket
import time
from concurrent.futures import ThreadPoolExecutor
def blocking_way():
sock = socket.socket()
# blocking
sock.connect(('example.com', 80))
request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'
sock.send(request.encode('ascii'))
response = b''
chunk = sock.recv(4096)
while chunk:
response += chunk
# blocking
chunk = sock.recv(4096)
return response
def multi_thread_way():
with ThreadPoolExecutor(max_workers=10) as executor:
futures = {executor.submit(blocking_way) for i in range(10)}
return len([future.result() for future in futures])
if __name__ == '__main__':
elapsed_times = 0
for _ in range(10):
start = time.time()
multi_thread_way()
elapsed_time = time.time() - start
elapsed_times += elapsed_time
print(f"elapsed_time: {(elapsed_time):.2f}[sec]")
print(f"mean_elapsed_time: {(elapsed_times/10):.2f}[sec]")
実行結果:
elapsed_time: 0.31[sec]
elapsed_time: 0.32[sec]
elapsed_time: 0.30[sec]
elapsed_time: 0.31[sec]
elapsed_time: 0.31[sec]
elapsed_time: 0.26[sec]
elapsed_time: 0.30[sec]
elapsed_time: 0.31[sec]
elapsed_time: 0.30[sec]
elapsed_time: 0.27[sec]
mean_elapsed_time: 0.30[sec]
10回の平均時間は0.30秒です。予想通り、マルチプロセスより速かったですね。スレッドにもコンテキストスイッチは存在しますが、プロセスと比べると大分軽いものになっています。これで、マルチスレッドはマルチプロセスのプロセス切り替えが遅いという問題を解決したようで、しかも同時処理できるタスクの規模もマルチプロセスの数百個から数千個に増加しました。
しかし、マルチスレッドにも問題があります。まず、Pythonのマルチスレッドは「GIL」の存在で、マルチコアCPUのアドバンテージを利用できないという問題があります。1つのPythonプロセス内で、ある時刻において1つのスレッドだけのアクティブ状態が許容されます。なら、なぜマルチスレッドはマルチプロセスよりも速かったでしょう。
理由はsock.connect()
、sock.recv()
のようなブロッキングするシステムコールを呼び出す時、現在のスレッドはGILを解放し、他のスレッドに実行できるチャンスを与えるからです。ただし、シングルスレッドの中でしたら、ブロッキングするシステムコールはそのままブロッキングします。
豆知識:
Pythonのtime.sleep
はブロッキングする処理ですが、マルチスレッドプログラミングでしたら、time.sleep()
は他のスレッドをブロッキングしません。
GIL以外に、マルチスレッドの共通した問題点もあります。スレッドはOSにスケジューリングされ、その「スケジューリングの戦略(scheduling strategy)」は「プリエンプション」で、同じ優先度のスレッドは均等的なチャンスで実行されるのが保証されます。プリエンプションは先取りの戦略なので、次の時刻にどのスレッドが実行されるのか、どのコードが実行されるのかは予測できず、「競合状態」になることがあります。
例えば、スクレイピングのワーカースレッドはタスクキューから次にスクレイピングするURLをpollする時、もし複数のスレッドが同時にpollしにきたら、どれに渡すかが問題になります。そこで、同じタスクが複数回実行されないようにロックや同期キューが必要になってきます。
また、マルチスレッドは数百から数千のマルチタスクを同時処理できますが、大規模かつ高頻度のウェブシステムに対してはまだ力不足です。もちろん、マルチスレッドの最大な問題点はやはり競合状態です。
3-4. 非同期ノンブロッキング方式
ようやく、ノンブロッキング方式まで来ました。まずは、一番原始的なノンブロッキングはどう作動するのかを見てみましょう。
import socket
import time
def nonblocking_way():
sock = socket.socket()
sock.setblocking(False)
# socketはノンブロッキング接続リクエストを送信すると、OSはエラーを起こすため
try:
sock.connect(('example.com', 80))
except BlockingIOError:
pass
request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'
data = request.encode('ascii')
# socketの接続がいつ確立されるか予測できないため、送信を繰り返す
while 1:
try:
sock.send(data)
break
except OSError:
pass
response = b''
# レスポンスがいつ読み取れるか予測できないため、受信を繰り返す
while 1:
try:
chunk = sock.recv(4096)
while chunk:
response += chunk
# blocking
chunk = sock.recv(4096)
break
except OSError:
pass
return response
def sync_way():
res = []
for i in range(10):
res.append(nonblocking_way())
return len(res)
if __name__ == '__main__':
elapsed_times = 0
for _ in range(10):
start = time.time()
sync_way()
elapsed_time = time.time() - start
elapsed_times += elapsed_time
print(f"elapsed_time: {(elapsed_time):.2f}[sec]")
print(f"mean_elapsed_time: {(elapsed_times/10):.2f}[sec]")
実行結果:
elapsed_time: 2.71[sec]
elapsed_time: 2.82[sec]
elapsed_time: 2.73[sec]
elapsed_time: 2.69[sec]
elapsed_time: 2.82[sec]
elapsed_time: 2.72[sec]
elapsed_time: 2.51[sec]
elapsed_time: 2.65[sec]
elapsed_time: 2.75[sec]
elapsed_time: 3.50[sec]
mean_elapsed_time: 2.79[sec]
10回の平均時間は2.79秒です。騙された気分ですね。同期ブロッキングと同じレベルの計算時間なのに、コードが余計に複雑になっています。ノンブロッキングなんかいらないと思うかもしれません。
まず、上記のコードでは、sock.setblocking(False)
で、OSにsocket
のブロッキングするシステムコールをノンブロッキングに変更するよう指示を出しました。前述のように、ノンブロッキングは1つのことをやるときに、自分を呼び出したプログラムが他のことをやるのを邪魔しないことです。上記のコードは、sock.connect()
とsock.recv()
は実行後、確かにブロッキングしなくなりました。
そして、コードが複雑になったのもブロッキングしなくなったからです。connect()
が呼び出されると、OSはまずエラーを起こすため、ここでtry
でキャッチする必要があります。そして、ここではブロッキングせずに、すぐに下のコードへ移ります。
while
文で繰り返し、send()
を実行するのは、connect()
がノンブロッキングになり、いつ接続が確立されるのか分からないため、試し続けるしかないからです。そして、send()
が実行されても、レスポンスがいつ来るか分からないため、recv()
の呼び出しも繰り返して実行しています。
connect()
とrecv()
はメインプログラムをブロッキングしなくなったが、CPUのあまった時間は有効に利用できてないです。その時間はただwhile
ループでsocket
の読み書きの繰り返しやエラー処理などに費やしていました。
そして、10個のダウンロードは依然のまま順番に実行されるため、トータルの計算時間は同期ブロッキング方式と変わらないわけです。
3-5. 非同期ノンブロッキン方式の改良
3-5-1. epoll
もしOS側はノンブロッキングの各呼び出しの用意ができているのかのチェックをしてくれるなら、アプリケーション側でループで待ったり判断したりしなくて済むので、空いた時間を他の処理に回すことで効率をアップできますね。
そのため、OSはI/O状態の変更をイベントとしてカプセル化してくれました。例えば、読み取り可能イベント、書き込み可能イベントなどがあります。そして、アプリケーションがイベント通知を受け取れるようにシステムモジュールも提供してくれました。そのモジュールはselect
です。アプリケーションはselect
を通して、「ファイル記述子」やコールバック関数を登録できます。ファイル記述子の状態が変化したら、select
は事前に登録されたコールバック関数を呼び出します。このやり方をI/O多重化(I/O multiplexing)と言います。
select
はアルゴリズムの効率が悪いため、のちにpoll
として改良されました。さらに、BSDカーネルはkqueue
モジュールに、Linuxカーネルはepoll
モジュールに改良しました。この4つのモジュールの機能は同じで、利用できるAPIもほぼ一緒です。kqueue
とepoll
は大量なファイル記述子を処理する時の効率が他の2つのモジュールより良いという違いがあります。
Linuxサーバーが普及しているため、epoll
モジュールはよく耳にしますね。ファイル記述子の数をselect・poll
はepoll
はepoll
はリスニングする全てのファイル記述子を、1個の特殊なファイル記述子でマウントします。このファイル記述子はマルチプロセス・スレッドで共有できるものです。
3-5-2. コールバック
I/OイベントのリスニングをOSに任せました。I/O状態が変化した時(例えばsocket接続が確立されデータを送信できるようになった時)、OSは次に何をすれば良いでしょう。それはコールバックです。
ここでは、データの送信と読み込みをそれぞれ独立した関数にカプセル化する必要があります。epoll
はアプリケーションの代わりにsocket
状態をリスニングする時に、epoll
に「socket
状態が書き込み可能な状態(接続が確立された)になったら、HTTPリクエスト送信関数を呼び出してください。socket
状態が読み取り可能な状態(クライアントがレスポンスを受信した)になったら、レスポンス処理関数を呼び出してください」と教える必要があります。
epoll
とコールバックを利用して、スクレイピングのコードをリファクタリングします。
import socket
import time
from selectors import DefaultSelector, EVENT_WRITE, EVENT_READ
class Crawler:
def __init__(self, path):
self.path = path
self.sock = None
self.response = b''
def fetch(self):
self.sock = socket.socket()
self.sock.setblocking(False)
try:
self.sock.connect(('example.com', 80))
except BlockingIOError:
pass
selector.register(self.sock.fileno(), EVENT_WRITE, self.connected)
def connected(self, key, mask):
selector.unregister(key.fd)
get = f'GET {self.path} HTTP/1.0\r\nHost: example.com\r\n\r\n'
self.sock.send(get.encode('ascii'))
selector.register(key.fd, EVENT_READ, self.read_response)
def read_response(self, key, mask):
global stopped
chunk = self.sock.recv(4096)
if chunk:
self.response += chunk
else:
selector.unregister(key.fd)
paths_todo.remove(self.path)
if not paths_todo:
stopped = True
前と違うのは、同じページではなく10個の違うページをダウンロードするところです。ダウンロードするページパスの集合paths_todo
は後で定義します。ここでは、改善したところを見てみましょう。
まず、send()
とrecv()
の2つのループが消えました。
続いて、selectors
モジュールを導入し、DefaultSelector
インスタンスを作成しました。Python標準ライブラリーのselectors
はselect/poll/epoll/kqueue
をカプセル化したものです。DefaultSelector
はOSによって一番良いモジュールを選んでくれます。Linux 2.5.44以上のバージョンは全部epoll
になっています。
そして、socket
の書き込み可能イベントEVENT_WRITE
と読み取り可能イベントEVENT_READ
が発生した後に処理するコールバック関数も登録しました。
コードの構造が綺麗になり、ブロッキングの通知などもOSに任せました。しかし、10個違うページをダウンロードするには、10個のCrawler
インスタンスが必要になり、20個のイベントが発生します。どうやってselector
から今発生したイベントを取得して、対応するコールバックを実行するでしょう。
3-5-3. イベントループ
上記の問題を解決するには、古いやり方を採用するしかないです。つまり、ループです。selector
モジュールにアクセスし、どのイベントが発生したか、どのコールバックを呼び出すべきかを知らせてくれるまで待ちます。この待ちのイベント通知ループをイベントループ(event loop)と言います。
def loop():
while not stopped:
# 何らかのイベントが発生するまでブロッキングする
events = selector.select()
for event_key, event_mask in events:
callback = event_key.data
callback(event_key, event_mask)
上記のコードで、stopped
というグローバル変数でイベントループがいつ止まるかをコントロールします。paths_todo
が全部消費されたら、stopped
をTrue
に変更します。
そして、selector.select()
はブロッキング呼び出しです。ここでイベントが発生しないと、アプリケーションは処理できるものがないため、イベントが発生するまでブロッキングする必要があります。予想できますが、1つのWebページだけダウンロードするとき、connect()
の後に、send()
そしてrecv()
ができるようになるため、同期ブロッキング方式と処理の効率は同じです。理由はconnect()
やrecv()
でブロッキングしなくても、select()
ではブロッキングするからです。
なので、selector
機能(ここからはepoll/kqueue
と呼ぶ)は大規模な並行アクセスを解決するためにデザインされたものです。システムの中に大量なノンブロッキング呼び出しがあり、ほぼランダムにイベントを生成できる状態でこそ、selector
機能は最大な威力を発揮できます。
下のコードは10個のダウンロードタスクを作成し、イベントループを起動したものです。
if __name__ == '__main__':
elapsed_times = 0
for _ in range(10):
selector = DefaultSelector()
stopped = False
paths_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}
start = time.time()
for path in paths_todo:
crawler = Crawler(path)
crawler.fetch()
loop()
elapsed_time = time.time() - start
elapsed_times += elapsed_time
print(f"elapsed_time: {(elapsed_time):.2f}[sec]")
print(f"mean_elapsed_time: {(elapsed_times/10):.2f}[sec]")
実行結果:
elapsed_time: 0.29[sec]
elapsed_time: 0.27[sec]
elapsed_time: 0.24[sec]
elapsed_time: 0.27[sec]
elapsed_time: 0.29[sec]
elapsed_time: 0.29[sec]
elapsed_time: 0.32[sec]
elapsed_time: 0.30[sec]
elapsed_time: 0.30[sec]
elapsed_time: 0.34[sec]
mean_elapsed_time: 0.29[sec]
10回の平均時間は0.29秒です。強いですね。シングルスレッドの中で、「イベントループ+コールバック」で、10個のページの同時ダウンロードという問題を解決しました。これこそ、非同期プログラミングです。for
ループがあり、順番的にCrawler
インスタンスを作成し、fetch
メソッドを呼び出ししているように見えますが、fetch
メソッドの処理はconnect()
とイベントの登録だけです。そして、実行時間から見れば、明らかにダウンロードタスクが同時に行われました。
上記コードの非同期処理の手順:
-
Crawler
インスタンスの作成 -
fetch
メソッドを呼び出し、socket
の接続を作りとselector
に書き込み可能イベントを登録する -
fetch
にはブロッキング処理がないため、すぐに返す - 上記のステップを10回繰り返し、10個のダウンロードタスクを全部イベントループに追加
- イベントループを起動し、最初のループに入り、イベントのリスニングにブロッキングする
- あるダウンロードタスクの
EVENT_WRITE
が発生したら、そのconnected
メソッドをコールバックし、最初のループを終了する - 2ラウンド目のループに入り、あるダウンロードタスクにイベントが発生したら、そのコールバック関数を実行する;この時点でどのイベントが発生するかは推測不可能になり、前回の
connected
の中のEVENT_READ
が発生するかもしれないし、他のタスクのEVENT_WRITE
が発生する可能性もあります(この時、1つのタスクにブロッキングした時間は他のタスクの実行に利用される) - 全てのダウンロードタスクが実行されるまでループが続く
- ループを終了し、プログラム全体を終了する
3-5-4. まとめ
同期ブロッキング方式から非同期ノンブロッキング方式まで見てきました。そして、シングルスレッドの中で、複数のネットワークI/Oブロッキング型タスクを並行処理する黒魔術もできるようになりました。マルチスレッドと比べると、スレッドの切り替えすらなくなりました。コールバックの実行は関数呼び出しで、スレッドのスタック内で完結します。しかも、性能も優れています。シングルサーバーで同時処理できるタスク数も数万から数十万といった規模になりました。
一部のプログラミング言語は、非同期プログラミングに対するサポートはここまでです。エンジニアは直接epoll
を使って、イベントやコールバックの登録、イベントループの管理、そして、コールバックの設計に多くの時間を費やす必要があります。
ここまでの内容を見ると分かると思いますが、どの言語を使っても、非同期プログラミングをするなら、上記の「イベントループ+コールバック」パターンから逃げられないです。ただし、epoll
を使ってないかもしれないし、while
ループではない可能性もあります。しかし、どれも「後で教えるよ」というモデルの非同期方式になります。
では、どうして一部の非同期プログラミングでは、コールバックパターンが見られてないでしょうか。これについてはこれから見ていきます。Pythonの非同期プログラミングの重役である、コルーチンについてはまだ何も話していないからですね。
4. Pythonの非同期I/Oに対する改良の道
ここからはPythonの非同期プログラミングのエコシステムはどうやって前述の「イベントループ+コールバック」パターンを受け継いだのかを説明し、そこからどのようにasyncio
というネイティブコルーチンパターンに進化したのかを見ていきましょう。
4-1. コールバック地獄
3. 非同期I/Oの進化の道で、シングルスレッドで非同期プログラミングを実現した「イベントループ+コールバック」の基本構造について見てきました。確かに、「イベントループ+コールバック」は大きくプログラムの効率を上げることができます。しかし、まだまだ問題解決ではないです。実際のプロジェクトは非常に複雑になるので、以下の問題を考えなくてはならないです。
- コールバックが正常に実行されなかった場合どうする
- どのようにコールバックの中にコールバックを組み込み、多重構造にする
- 多重構造にしたら、ある層でエラーが起きたらどうなる
- あるデータを全てのコールバックに処理させる必要がある時はどうする
- ...
実際のプロジェクトでは、これらの問題は避けられないです。そして、問題の背後にコールバックパターンの欠点が隠れています。
- 多重コールバックにすると、可読性が著しく低下する
def callback_1():
# 処理
def callback_2():
# 処理
def callback_3():
# 処理
def callback_4():
# 処理
def callback_5():
#処理
async_function(callback_5)
async_function(callback_4)
async_function(callback_3)
async_function(callback_2)
async_function(callback_1)
(Lisp信者はご容赦ください)
- コードの構造をぶっ壊すことになる
同期プログラミングのコードを書く時、関連のある処理は普通に上から下へと実行されます。
do_a()
do_b()
もし、do_b()
はdo_a()
の結果に依存し、そしてdo_a()
は非同期呼び出しの場合、do_a()
の結果がいつ返ってくるか分からないため、後続の処理をコールバックの形式でdo_a()
に渡す必要があります。こうすることで、do_a()
が完了してからdo_b()
の実行に入ることを保証できます。
do_a(do_b())
そして、長い処理フローを全部非同期処理にすると、こうなります。
do_a(do_b(do_c(do_d(do_e(do_f(...))))))
上記のスタイルは「コールバック地獄スタイル」と言います。しかし、主な問題点は見た目ではなく、本来の上から下への構造を外から内へと変更する必要があるところです。まずはdo_a()
、そしてdo_b()
、続いてdo_c()
、…、一番内側にあるdo_f()
へと実行していくことになります。同期処理で、do_a()
の後にdo_b()
というのは、スレッドの命令ポインタがフローをコントロールしてくれます。しかし、コールバックパターンになると、フローのコントロールはエンジニアが注意しながら配置することになります。
- 状態の共有や管理が難しい
3-1の同期バージョンのsock
オブジェクトは上から下へとずっと使い回されるのに対して、3-5のコールバックバージョンはCrawler
クラスをインスタンス化して、sock
オブジェクトをself
に保存する必要があります。オブジェクト指向的なプログラミングスタイルを採用しないと、共有する必要のある状態をバトンタッチみたいにそれぞれのコールバック関数に渡す必要が出てきます。そして、複数の非同期呼び出しの間に、どの状態を共有しなればならないかは事前に計画して、入念にデザインする必要があります。
- エラー処理が難しい
一連のコールバックは呼び出しのチェーン(「メソッドチェーン」)を構成します。例えば、do_a()
からdo_f()
というようなチェーンがあります。もし、do_d()
がエラーを起こしたらどうなると思いますか?チェーンが切れて、バトンタッチしてた状態も失います。そして、「スタックトレース」をぶっ壊すことになります。例えば、do_d()
がエラーを起こし、do_c()
の内部でdo_d()
の呼び出しがエラーになったため、do_c()
自体もエラーを起こすことになります。同様にdo_b()
もdo_a()
もエラーを起こすことになり、結局エラーログは「do_a()
の呼び出しがエラーを起こした」とだけ報告することになります。しかし、実際エラーを起こしたのはdo_d()
です。これを防ぐために、エラーを全部キャッチして、関数の戻り値としてデータを返させる必要があります。そして、全てのコールバック関数は前の関数の戻り値をチェックする必要があり、「エラーの隠蔽」を防ぎます。
というわけで、可読性はただ見栄えの問題だとしても、スタックトレースをぶっ壊すのと状態の共有や管理が難しいといった2つの欠点はコールバックベースの非同期プログラミングの異常な難しさの原因になります。あらゆるプログラミング言語はこの問題を解決しようとしています。そのおかげで、「Promise」、「コルーチン」といったソリューションが生まれたわけです。
4-2. 課題
「非同期タスクがいつ完了するか」、「非同期呼び出しの戻り値をどう処理するか」といった非同期プログラミングの難点について、「イベントループ+コールバック」パターンで解決できました。しかし、コールバックはプログラムを複雑にしてしまいます。この欠点を回避する手段を考える前に、まず本質的なところをはっきりする必要があります。なぜ、コールバックは必須なのでしょうか?そして、そもそも、欠点の1つとされる状態の共有や管理は何のためでしょうか?
状態の共有や管理が必要なのは、プログラムは自分は何をした、今何している、これから何をするというのを知る必要があるからです。言い換えると、プログラムは自身の現在の状態を把握する必要があり、その状態をそれぞれのコールバックにバトンタッチして、保持し続ける必要もあります。
複数のコールバック間の状態の管理が難しいです。ならば、それぞれのコールバックに自分の状態だけを管理させるのはどうでしょう?呼び出しのチェーンは、エラーハンドルを難しくします。ならば、呼び出しのチェーンを使わないのはどうでしょう?しかし、呼び出しのチェーンを使わないなら、呼び出された関数は前の関数が完了しているかどうかをどう知るのでしょうか?では、コールバックに次のコールバックに通知させるのはどうでしょう?そもそも、コールバックというのは、保留中(pending)のタスクという見方もできますね。
タスク間の相互通知、タスク毎に自分の状態を持たせる、これはまさに古から存在したプログラミング作法「協調的マルチタスク」です。しかし、シングルスレッドの中にスケジューリングする必要があります。ここからは、スタックフレームを持ち、自身の状態を簡単に知ることができる「コルーチン」の出番です。コールチン間はもちろん通知し合うことも可能です。
4-3. コルーチン
- コルーチン(co-routine)、言い換えると協調的なルーチン
コルーチンは「サブルーチン」を一般化したものとされています。コルーチンのスケジューリング戦略は「ノンプリエンプティブ」で、複数のエントリーで、処理の中断と再開をコントロールできます。
サブルーチンはプログラミング言語が定義した呼び出し可能なコードブロックです。言い換えると、ある機能を実現するためにパックされた一連の命令ですね。一般的なプログラミング言語では、関数やメソッドといった構造でサブルーチンを実現しています。
4-4. ジェネレーターベースのコルーチン
Pythonの中に特殊なオブジェクト「ジェネレーター」があります。ジェネレーターの特徴はコルーチンと似ていて、イテレーションの間に、中断でき、そして次のイテレーションまで今までの状態を失うことがありません。
ジェネレーターで簡単なコルーチンを実現するため、Python 2.5でジェネレーターの機能が強化(PEP 342)されました。この機能強化の提案タイトルは「Coroutines via Enhanced Generators」です。PEP 342のおかげで、ジェネレーターはyield
で実行を中断し、データを返すことができるようになりました。そして、send
でジェネレーターにデータを送ることもでき、throw
でジェネレーター内にエラーを起こさせ、終了させることも可能になりました。
次に、ジェネレーターベースのコルーチンで、スクレイピングのプログラムをリファクタリングします。
4-4-1. Futureオブジェクト
コールバック方式を辞めると、非同期呼び出しの結果をどうやって知るのでしょうか?ここでは、まずオブジェクトを定義します。非同期呼び出しの結果が返ってきたら、その中に保存します。このオブジェクトをFuture
オブジェクトと呼びます。
import socket
import time
from selectors import DefaultSelector, EVENT_WRITE, EVENT_READ
class Future:
def __init__(self):
self.result = None
self._callbacks = []
def add_done_callback(self, fn):
self._callbacks.append(fn)
def set_result(self, result):
self.result = result
for fn in self._callbacks:
fn(self)
Future
オブジェクトはresult
インスタンス変数を持ち、将来の実行結果を蓄積します。そして、set_result
メソッドはresult
を設置するもので、result
に値をバインドしたら、Future
オブジェクトに事前に追加されたコールバック関数を実行します。コールバック関数は、add_done_callback()
メソッドで追加できます。
コールバック方式を辞めると言いませんでしたっけ?慌てないでください。ここで説明したように、非同期プログラミングをするなら、「イベントループ+コールバック」パターンから逃げられないです。そして、ここのコールバックは前のものと少し違います。
4-4-2. Crawlerをリファクタリング
とにかく、将来のデータを表すFuture
オブジェクトを作りました。Future
オブジェクトでスクレイピングのコードをリファクタリングしてみます。
class Crawler:
def __init__(self, path):
self.path = path
self.response = b''
def fetch(self):
sock = socket.socket()
sock.setblocking(False)
try:
sock.connect(('example.com', 80))
except BlockingIOError:
pass
f = Future()
def on_connected():
f.set_result(None)
# fileno()メソッドはソケットのファイル記述子を短い整数型で返します
selector.register(sock.fileno(), EVENT_WRITE, on_connected)
yield f
selector.unregister(sock.fileno())
get = f'GET {self.path} HTTP/1.0\r\nHost: example.com\r\n\r\n'
sock.send(get.encode('ascii'))
global stopped
while True:
f = Future()
def on_readable():
f.set_result(sock.recv(4096))
selector.register(sock.fileno(), EVENT_READ, on_readable)
# sendされたresultを受け取る
chunk = yield f
selector.unregister(sock.fileno())
if chunk:
self.response += chunk
else:
paths_todo.remove(self.path)
if not paths_todo:
stopped = True
break
前のコールバックバージョンと比べると、かなり違いが出ています。fetch
メソッドにyield
表現を使い、ジェネレーターにしました。ジェネレーターはnext()
でイテレーションを1回回すかsend(None)
で起動でき、yield
まで行くと中断します。では、fetch
ジェネレーターはどうやって再開するのでしょうか?
4-4-3. Taskオブジェクト
上記の問題を解決するために、1つのルールに従う必要があります。それは、「単一責任の原則」 です。そのため、ここでは、ジェネレーターの再開や状態の管理という役割を果すものを作ります。名前はTask
にします。
class Task:
def __init__(self, coro):
# コルーチンオブジェクト
self.coro = coro
f = Future()
f.set_result(None)
self.step(f)
def step(self, future):
try:
# sendすると、ジェネレーターは次のyieldまで実行される
# next_futureはyieldが返したオブジェクト
next_future = self.coro.send(future.result)
except StopIteration:
return
next_future.add_done_callback(self.step)
上記コードのTask
オブジェクトはcoro
コルーチンオブジェクトをパックします。管理するタスクは、保留中(pending)のタスク(コルーチン)のため、ここのcoro
はfetch
ジェネレーターになるわけです。そして、step
メソッドがあり、初期化の時にまず1回実行されます。step
メソッドはジェネレーターのsend()
メソッドを呼び出し、初期化の時はsend(None)
になるため、coro
すなわちfetch()
の初回イテレーションが回されます。
send()
完了後、次のfuture
が得られて、add_done_callback
を使って次のfuture
にstep()
というコールバック関数を追加します。
次に、fetch()
ジェネレーターを見てみましょう。リクエストの送信やレスポンスの受信などのビジネスロジックは内部で完結しています。そして、selector
に登録するコールバック関数もシンプルになりました。2つのyield
は対応するfuture
を返し、Task.step()
内で受け取ります。これで、Task
、Future
、Coroutine
をうまく繋ぎ合わせることができました。
Task
オブジェクトを初期化し、fetch()
が最初のyield
まで実行されます。では、どうやって再開するのでしょうか?
4-4-4. イベントループ
イベントループの再登場です。最初のyieldまで来たら、登録されたEVENT_WRITE
が発生するのを待ちます。イベントループは心拍みたいに、脈動し始めれば、ずっと動き続きます。
def loop():
while not stopped:
# 何らかのイベントが発生するまでブロッキングする
events = selector.select()
for event_key, event_mask in events:
callback = event_key.data
callback()
if __name__ == '__main__':
elapsed_times = 0
for _ in range(10):
selector = DefaultSelector()
stopped = False
paths_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}
start = time.time()
for path in paths_todo:
crawler = Crawler(path)
Task(crawler.fetch())
loop()
elapsed_time = time.time() - start
elapsed_times += elapsed_time
print(f"elapsed_time: {(elapsed_time):.2f}[sec]")
print(f"mean_elapsed_time: {(elapsed_times/10):.2f}[sec]")
実行結果:
elapsed_time: 0.30[sec]
elapsed_time: 0.28[sec]
elapsed_time: 0.28[sec]
elapsed_time: 0.29[sec]
elapsed_time: 0.28[sec]
elapsed_time: 0.25[sec]
elapsed_time: 0.26[sec]
elapsed_time: 0.26[sec]
elapsed_time: 0.28[sec]
elapsed_time: 0.26[sec]
mean_elapsed_time: 0.27[sec]
10回の平均時間は0.27秒です。今回のloop
は前と少し違います。callback()
はevent_keyとevent_mask
を受け取らなくなりました。つまり、ここのコールバックは誰がイベントを発生させたのを知る必要がなく、fetch()
と合わせて見れば分かりますが、コールバックはただfuture
にset_result()
で値を設置すれば良いです。そして、どのfuture
なのも知る必要がなく、コルーチンは自身の状態を保存でき、自分のfuture
さえ分かれば問題ないです。どの値を設置するのも気にする必要がなく、コルーチンは全部処理してくれます。
4-4-5. 「ジェネレーター・コルーチンスタイル VS コールバックスタイル」のまとめ
コールバックスタイル:
- 呼び出しのチェーンが存在する
- リクエストとレスポンスは2つのコールバックに分けなければならない。コードの構造を破壊した
- コールバック間の状態保持をエンジニアが管理する必要がある
ジェネレーター・コルーチンスタイル:
- 呼び出しのチェーンが存在しない
-
selector
のコールバックはfuture
に値を設置するだけで、ビジネスロジックから切り離せた -
loop
内のcallback()
は誰がイベントを発生させたのを知る必要がなくなった - 同期処理に近い構造になった
- 各コルーチン間の状態保持をエンジニアが管理する必要がなくなった(例えば、自分の
sock
はどれかなど)
4-4-6. コードを更にリファクタリングする
コードは少し読みにくいですね。ここでもし、fetch
の耐欠陥性、機能を更に高度にしてくださいと言われたらどうすれば良いでしょう。また、技術ロジック(socket
関連)とビジネスロジック(リクエストとレスポンスの処理)が混在していて良くないですね。
-
socket
接続の作成は抽象化(関数・メソッド化)できる - レスポンスの読み込みループも抽象化できる
- ループ内の
socket.recv()
も抽象化できる
しかし、これらのところにyield
が存在し、抽象化するなら、ジェネレーターにする必要があります。それに、fetch()
自身もジェネレーターで、ジェネレーターの中でジェネレーターを弄るのは、コードを更に複雑にするかもしれません。
Pythonの設計者たちもこの問題に気づき、yield from
というジェネレーターの中でジェネレーターを弄るおもちゃを用意しくれました。
4-5. yield fromでジェネレーターベースのコルーチンを改良
4-5-1. yield from文法の紹介
yield from
はPython 3.3から導入された文法(PEP 380)です。PEP 380は主に、ジェネレーターの中でジェネレーターを弄る時の不便を解消するためのもので、2つの機能があります。
その1つは、イテレーションを回してサブジェネレーターをyield
する必要がなくなり、直接yield from
できるという機能です。以下の2種類のジェネレーターは機能として同等です。
def gen_1():
sub_gen = range(10)
yield from sub_gen
def gen_2():
subgen = range(10)
for item in subgen:
yield item
もう1つは、サブジェネレーターとメインジェネレーターの間に相互通信できるチャンネルを開けてくれる機能です。
def gen():
yield from sub_gen()
def sub_gen():
while 1:
x = yield
yield x + 1
def main():
g = gen()
next(g) # 最初のyieldまで実行させる
retval = g.send(1) # gen()にデータを送るように見えるが、実はsub_gen()に送っている
print(retval) # sub_gen()から計算済みの2が出力される
g.throw(StopIteration) # sub_gen()のところでエラーを起こす
上記のコードはyield from
の相互通信機能を示したものです。yield from
はgen()
の内部に、sub_gen()
とmain()
の間に通信チャンネルを開けてくれています。main()
から直接データ1
をsub_gen()
に送ることができ、sub_gen()
からも直接計算済みの値2
をmain()
に返すことができます。また、main()
から直接sub_gen()
にエラーを送ることで、sub_gen()
を終了させることもできます。
ちなみに、yield from
はyield from <generator>
だけではなく、yield from <iterable>
も可能です。
4-5-2. リファクタリング
socket
接続を抽象化
def connect(sock, address):
f = Future()
sock.setblocking(False)
try:
sock.connect(address)
except BlockingIOError:
pass
def on_connected():
f.set_result(None)
selector.register(sock.fileno(), EVENT_WRITE, on_connected)
yield from f
selector.unregister(sock.fileno())
レスポンスの読み込みを抽象化
def read(sock):
f = Future()
def on_readable():
f.set_result(sock.recv(4096))
selector.register(sock.fileno(), EVENT_READ, on_readable)
chunk = yield from f
selector.unregister(sock.fileno())
return chunk
def read_all(sock):
response = []
chunk = yield from read(sock)
while chunk:
response.append(chunk)
chunk = yield from read(sock)
return b''.join(response)
Crawler
のリファクタリング
class Crawler:
def __init__(self, path):
self.path = path
self.response = b''
def fetch(self):
global stopped
sock = socket.socket()
yield from connect(sock, ('example.com', 80))
get = f'GET {self.path} HTTP/1.0\r\nHost: example.com\r\n\r\n'
sock.send(get.encode('ascii'))
self.response = yield from read_all(sock)
paths_todo.remove(self.path)
if not paths_todo:
stopped = True
ここまでのコードは特に問題ありません。再利用できる部分は関数として抽象化しました。サブジェネレーターの値もyield from
で取得できるようにしました。ただし、1つ注意すべきは、future
オブジェクトを返すときに、yield
ではなく、yield from
を使用している点です。yield
は普通のPythonオブジェクトに適用できますが、yield from
はできません。ここで、Future
を改造して、iterable
オブジェクトにする必要があります。
class Future:
def __init__(self):
self.result = None
self._callbacks = []
def add_done_callback(self, fn):
self._callbacks.append(fn)
def set_result(self, result):
self.result = result
for fn in self._callbacks:
fn(self)
def __iter__(self):
yield self
return self.result
__iter__
を追加しただけです。もちろん、必ずしもyield from
を使う必要はないですが、yield
のままでも問題ありません。しかし、ジェネレーターベースのコルーチンをしているのか、それともジェネレーターだけなのかを区別するために、使い分けをしたほうが良いです。そのため、Python 3.3からyield from
が導入された後、yield
でコルーチンを作るのは非推奨になったわけです。yield from
の相互通信機能を利用して、コルーチン間にデータを自由自在に送り合えるメリットもあります。
4-5-3. yield fromによるコルーチン改良のまとめ
yield from
によるコルーチンの改良で、コードの抽象レベルを高められ、ビジネスロジック関連を更にシンプル化できました。相互通信機能により、コルーチン間のデータのやりとりが簡単になりました。これで、Python非同期プログラミングは大きな進歩を遂げました。
そして、Python言語の開発者たちもyield from
をフル活用しました。Guidoさんが主導したPython非同期プログラミングフレームワークTulip
も猛スピードで進化し、Python 3.4で名前をasyncio
に変更して、標準ライブラリとして仮採用(on a provisional basis)しました。
4-6. asyncio
4-6-1. asyncioの紹介
asyncio
はPython 3.4から実験的に導入された非同期I/Oフレームワーク(PEP 3156)です。asyncio
はPythonにおいて、コルチーンによる非同期I/Oプログラミングのインフラとして提供されました。コアのコンポーネントとして、イベントループ(Event Loop)、コルーチン(Coroutine)、タスク(Task)、フューチャー(Future)とその他の補助的なモジュールから構成されています。
asyncio
が導入された時、@asyncio.coroutine
というデコレーターも提供されました。yield from
を使用した関数につけることで、コルーチンとしてマークすることができますが、強制使用は強いられていないです。
Python 3.4からyield from
の助力で、コルーチンが簡単に作れるようになりましたが、歴史的な問題もそうですが、人々はジェネレーターとコルーチンの区別と関係性が分かりません。そして、yield
とyield from
の違いも分かりません。この混乱状態は「Pythonの禅」のルールに背いています。
それで、Python 3.5から、Pythonの設計者たちは急ぎながら、async/await
文法(PEP 492)を追加し、コルーチンを明示的に支持している様子を見せました。これをネイティブコルーチンと呼びます。async/await
とyield from
の2つのコルーチンスタイルは内部実装が同じで、お互い互換性があります。
そして、Python 3.6から、asyncio
が正式的に標準ライブラリーに仲間入りしました。以上はCPythonにおいての非同期I/Oの進化の軌跡です。
4-6-2. asyncioとネイティブコルーチン
asyncio
とasync/await
文法の便利さを体験してみましょう。
import asyncio
import aiohttp
import time
async def fetch(session, url):
async with session.get(url) as response:
response = await response.read()
return response
async def fetch_many(host, paths_todo):
async with aiohttp.ClientSession(loop=loop) as session:
tasks = [fetch(session, host + path) for path in paths_todo]
responses = await asyncio.gather(*tasks)
return responses
if __name__ == '__main__':
loop = asyncio.get_event_loop()
host = 'http://example.com'
paths_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}
elapsed_times = 0
for _ in range(10):
start = time.time()
loop.run_until_complete(fetch_many(host, paths_todo))
elapsed_time = time.time() - start
elapsed_times += elapsed_time
print(f"elapsed_time: {(elapsed_time):.2f}[sec]")
print(f"mean_elapsed_time: {(elapsed_times/10):.2f}[sec]")
実行結果:
elapsed_time: 0.26[sec]
elapsed_time: 0.26[sec]
elapsed_time: 0.24[sec]
elapsed_time: 0.24[sec]
elapsed_time: 0.26[sec]
elapsed_time: 0.25[sec]
elapsed_time: 0.26[sec]
elapsed_time: 0.26[sec]
elapsed_time: 0.25[sec]
elapsed_time: 0.26[sec]
mean_elapsed_time: 0.25[sec]
ジェネレーターベースのコルーチンスタイルと比較すると、asyncio
スタイルは結構違いますね。
-
yield
やyield from
が消え、代わりにasync/await
が使われる - 自作の
loop()
が消え、代わりにasyncio.get_event_loop()
が使われる -
socket
に対する明示的な処理は必要なくなり、aiohttp
ライブライがやってくれる -
Future
やTask
の定義も消え、asyncio
が実装してくれている - 少ないコードで、エレガントな設計を実現できた
HTTPリクエストの送信とレスポンスの受信などを自分でsocket
を操作しない理由は、実際の業務で、うまくHTTPプロトコルを処理するのは極めて難しくて、機能が充実した非同期HTTPクライアントさえあれば、自力でする必要がなくなるからです。
同期ブロッキングバージョンのコードと比較すると:
- 非同期化
- コード数はほぼ一緒(
aiohttp
の導入で逆に少ない) - コードのロジックは同じぐらいシンプル
- 10倍ぐらいの性能アップ
まとめ
Pythonにおいての非同期プログラミングの発展から仕組みまで詳しく見てきました。最終的に、同期処理と同じぐらいシンプルなコードでN倍の効率向上を実現しました。しかも、コールバック地獄のような欠点もありません。
asyncio
の詳しい使い方、その長所と短所、そして、Pythonエコシステム内の他の非同期I/Oソリューション、asyncio
との区別などはまた他の記事で紹介します。
参考
A Web Crawler With asyncio Coroutines
Latency numbers every programmer should know
Discussion