ひとつのループで数百の接続をさばく

目標

逐次サーバーの行列待ちを自分で測り、同じ負荷を1つのループで受けさばく多重化サーバーを作って、2つの測定値を比べます。

なぜ重要なのか

接続が1つ話し始めるのが遅いと、逐次サーバーはそのあいだ何もできません。CPUは空いていてログも静かなので、指標だけを見ていると、塞がっているという事実自体が見えません。待つ場所を接続ごとに置かず、1か所にまとめればこの問題はなくなりますが、その代わり、部分読み取りと部分書き込み、関心イベント、アイドル整理を、プログラムが自分で管理する必要があります。このラボは、その管理のコストを1つずつ作ってみて、変える前と後を同じ物差しで測り、何が実際によくなったのかを、数字で残します。標準ライブラリだけを使い、インターネット接続やインストールは必要ありません。

ステップ

止まらないリッスンソケットを作る: /root/mux/server.pyにmake_listener(host, port, backlog=64)を作ってください。AF_INET・SOCK_STREAMのソケットでSO_REUSEADDRを有効にし、hostとportでbindしたあと、backlogでlistenして、setblocking(False)に変えたソケットを返します。port 0は、OSが空いているポートを選ぶことを許可します。hostが空でない文字列でない、portが0から65535までのintでない、backlogが1以上のintでない場合は、ValueErrorです。boolはintとして受け付けません。
行列待ちを数字で残す: python3 /opt/fixtures/mux/probe.py --target sequentialを実行してください。出力の最後にあるblockedとfastの2つの値を、/root/mux/report.txtにseq_blocked=<値>とseq_fast=<値>の2行で書きます。採点ツールが同じ測定をその場でもう一度実行して、2つの値を照合します。
1回起きたときに待ち行列を空にする: server.pyにaccept_all(listener)を追加してください。BlockingIOErrorが出るまでacceptを繰り返して、受け入れたソケットのlistを返します。待っている接続がなければ空のlistで、受け入れたソケットもsetblocking(False)に変えます。ConnectionAbortedErrorやConnectionResetErrorが出た接続は飛ばして、残りの待ち行列は受け取り続けます。
受け取った分だけ読み、残りは保管する: server.pyにConnクラスを追加してください。Conn(sock, now=0.0)は、inboxとoutboxを空のbytesに、closedをFalseに、last_activeをfloat(now)にします。on_readable()は、sock.recvを1回だけ呼び出し、受け取ったバイトをinboxの後ろにつなげたあとで、改行で終わった行だけ、改行を取ってlistで返します。未完成の断片はinboxに残します。recvがb""を返したらclosedをTrueにして空のlist、BlockingIOErrorは何も変えずに空のlist、ConnectionResetErrorはclosedをTrueにして空のlistです。長さ0の行も、1つの行です。
送った分だけ消す: Connにqueue(data)とwants_write属性とon_writable()を追加してください。queueはbytesだけを受け付けてoutboxの後ろにつなげ、bytesでなければValueErrorです。wants_writeは、outboxが空でないかどうかです。on_writableは、outboxが空ならsendなしで0を、そうでなければsock.sendを1回呼び出して、返ってきた数だけoutboxの先頭から切り取り、その数を返します。BlockingIOErrorは0で、outboxはそのまま残します。BrokenPipeErrorとConnectionResetErrorは、closedをTrueにして0です。
書き込みは送るものがあるときだけ見張る: server.pyにinterest(conn)を追加してください。connのoutboxが空ならselectors.EVENT_READだけを、残っていればselectors.EVENT_READとselectors.EVENT_WRITEをORで合わせた値を返します。
静かな接続を選ぶ: Connにtouch(now)を、server.pyにidle_keys(conns, now, idle_timeout)を追加してください。touchは、last_activeをfloat(now)に変えます。idle_keysは、キーからConnへのディクショナリを受け取り、now - last_activeがidle_timeout以上のキーだけを並べ替えてlistで返し、受け取ったディクショナリは変更しません。idle_timeoutが正の有限なintまたはfloatでなければValueErrorで、boolは受け付けません。
1つのループですべて受け付け、もう一度測って比べる: server.pyにserve(host, port, idle_timeout, ready=None, clock=None)を追加してください。make_listenerで待ち受けを始めたあと、readyがあれば、実際にbindされたポート番号で1回呼び出し、selectorsでリッスンソケットとすべての接続を一緒に見張りながら繰り返します。REQの後ろに値が付いた行を受け取ったら、OKに同じ値を付けた行を、その接続のoutboxに入れ、interestで関心イベントを更新し、closedの接続とidle_keysが選んだ接続は、関心リストから外して閉じます。イベントがなくてもアイドル検査が回るように、多重化の呼び出しの待ち時間に上限を置いてください。そのあと、python3 /opt/fixtures/mux/probe.py --target /root/mux/server.pyとpython3 /opt/fixtures/mux/fdcount.py /root/mux/server.pyを実行して、/root/mux/report.txtにmux_blocked=<値>、mux_fast=<値>、mux_fds=<値>の3行を追加してください。

参考

すべての関数は1つのファイル/root/mux/server.pyに置き、測定値は/root/mux/report.txtに書きます。mkdir -p /root/muxで、作業フォルダを先に作ってください。
測定ツールは/opt/fixtures/muxにあります。probe.pyは行列待ちを、fdcount.pyは接続200個のときのファイル記述子の数を測り、seqserver.pyは比べる基準となる逐次サーバーです。3つとも読んでかまいません。
採点ツールは、書き出した数字をその場でもう一度測ります。もっともらしい値を書くだけでは、合格しません。
例のうち、まだ実装していない関数は枠のまま残しますが、完成した関数は上書きしないでください。採点は、前のステップの契約もあわせて見ます。
よくあるミスが2つあります。sendが返した数を無視して、outboxをまるごと空にすることと、送るものがないのに書き込みの準備完了を見張り続けて、コア1つが100%に張り付くことです。
ラボのセッションが終わると、/rootのファイルは残りません。必要なコードは、終える前に別に保管してください。

止まらないリッスンソケットを作る

/root/mux/server.pyにmake_listener(host, port, backlog=64)を作ってください。AF_INET・SOCK_STREAMのソケットでSO_REUSEADDRを有効にし、hostとportでbindしたあと、backlogでlistenして、setblocking(False)に変えたソケットを返します。port 0は、OSが空いているポートを選ぶことを許可します。hostが空でない文字列でない、portが0から65535までのintでない、backlogが1以上のintでない場合は、ValueErrorです。boolはintとして受け付けません。

bindとlistenは、カーネルの状態を変えるだけで、相手を待ちません。プロセスを止めるのはacceptで、それを防ぐ1行が、ソケットのモードです。検査では、boolがintのサブタイプだという点を、別に扱う必要があります。

行列待ちを数字で残す

python3 /opt/fixtures/mux/probe.py --target sequentialを実行してください。出力の最後にあるblockedとfastの2つの値を、/root/mux/report.txtにseq_blocked=<値>とseq_fast=<値>の2行で書きます。採点ツールが同じ測定をその場でもう一度実行して、2つの値を照合します。

測定ツールは、遅いお客さん1人を先に付けて、速いお客さん数人を後ろに並べます。blockedは1秒を超えたお客さんの数、fastは0.2秒以内に終わったお客さんの数です。推測して書くと、測り直した値とずれます。出力をファイルに残しておくと、最後のステップで比べやすくなります。

1回起きたときに待ち行列を空にする

server.pyにaccept_all(listener)を追加してください。BlockingIOErrorが出るまでacceptを繰り返して、受け入れたソケットのlistを返します。待っている接続がなければ空のlistで、受け入れたソケットもsetblocking(False)に変えます。ConnectionAbortedErrorやConnectionResetErrorが出た接続は飛ばして、残りの待ち行列は受け取り続けます。

読み取りの準備完了の通知は、待っている接続が1つだと教えてくれません。いくつあるかを教えてくれる値もないので、もうないという合図が来るまで繰り返すのが、唯一の方法です。その合図は、エラーではなく、正常な終了条件です。

受け取った分だけ読み、残りは保管する

server.pyにConnクラスを追加してください。Conn(sock, now=0.0)は、inboxとoutboxを空のbytesに、closedをFalseに、last_activeをfloat(now)にします。on_readable()は、sock.recvを1回だけ呼び出し、受け取ったバイトをinboxの後ろにつなげたあとで、改行で終わった行だけ、改行を取ってlistで返します。未完成の断片はinboxに残します。recvがb""を返したらclosedをTrueにして空のlist、BlockingIOErrorは何も変えずに空のlist、ConnectionResetErrorはclosedをTrueにして空のlistです。長さ0の行も、1つの行です。

1回のrecvが1つのメッセージを返すという保証はありません。前の断片とつなげてはじめて、改行を探せますし、改行がないというのは、不正な入力ではなく、まだ全部来ていないという意味です。b""と空行は、値は似て見えても、意味が正反対です。

送った分だけ消す

Connにqueue(data)とwants_write属性とon_writable()を追加してください。queueはbytesだけを受け付けてoutboxの後ろにつなげ、bytesでなければValueErrorです。wants_writeは、outboxが空でないかどうかです。on_writableは、outboxが空ならsendなしで0を、そうでなければsock.sendを1回呼び出して、返ってきた数だけoutboxの先頭から切り取り、その数を返します。BlockingIOErrorは0で、outboxはそのまま残します。BrokenPipeErrorとConnectionResetErrorは、closedをTrueにして0です。

sendの戻り値は、今回の呼び出しが処理したバイト数です。短い応答でテストすると、たいてい1回で全部出ていくので、誤った実装も通ってしまいます。20バイトを入れて3だけ出ていったとき、outboxに何が残るべきかを、先に決めてください。

書き込みは送るものがあるときだけ見張る

server.pyにinterest(conn)を追加してください。connのoutboxが空ならselectors.EVENT_READだけを、残っていればselectors.EVENT_READとselectors.EVENT_WRITEをORで合わせた値を返します。

読み取りの準備完了と書き込みの準備完了は、真になる頻度がまったく違います。送信バッファーはほとんどの時間が空なので、書き込みの準備完了はほとんど常に真です。その事実が、この関数の条件を決めます。

静かな接続を選ぶ

Connにtouch(now)を、server.pyにidle_keys(conns, now, idle_timeout)を追加してください。touchは、last_activeをfloat(now)に変えます。idle_keysは、キーからConnへのディクショナリを受け取り、now - last_activeがidle_timeout以上のキーだけを並べ替えてlistで返し、受け取ったディクショナリは変更しません。idle_timeoutが正の有限なintまたはfloatでなければValueErrorで、boolは受け付けません。

選ぶ処理と切る処理を1つの関数に混ぜると、反復の途中でデータ構造が変わります。選ぶほうを純粋関数にしておけば、時計を渡すだけで境界条件をテストできます。経過が制限とちょうど同じときにどちらにするかを、先に決めてください。

1つのループですべて受け付け、もう一度測って比べる

server.pyにserve(host, port, idle_timeout, ready=None, clock=None)を追加してください。make_listenerで待ち受けを始めたあと、readyがあれば、実際にbindされたポート番号で1回呼び出し、selectorsでリッスンソケットとすべての接続を一緒に見張りながら繰り返します。REQの後ろに値が付いた行を受け取ったら、OKに同じ値を付けた行を、その接続のoutboxに入れ、interestで関心イベントを更新し、closedの接続とidle_keysが選んだ接続は、関心リストから外して閉じます。イベントがなくてもアイドル検査が回るように、多重化の呼び出しの待ち時間に上限を置いてください。そのあと、python3 /opt/fixtures/mux/probe.py --target /root/mux/server.pyとpython3 /opt/fixtures/mux/fdcount.py /root/mux/server.pyを実行して、/root/mux/report.txtにmux_blocked=<値>、mux_fast=<値>、mux_fds=<値>の3行を追加してください。

前のステップで作った関数を、そのままつなげればよいのです。ループが新しく決めるのは、順序だけです。受け入れ、読み、書き、関心イベントを計算し直し、切るものを切ります。fdsの値は実装に付随する数字なので、他人の答えと違うことがあり、なぜ200ではないのかを説明できなければなりません。