大文字印刷所で消えた返事を取り戻す

目標

送信方向だけを終了したクライアントに最後まで返信し、メモリと総寿命を制限する、多重化サーバーの構成要素を作ります。

なぜ重要なのか

EOFで両方向を閉じると、まだ送っていない応答が消えます。逆に無限に待つと、遅いお客さんが接続リソースを握り続けます。入力の終了、出力の残り、最終的な理由を分けて、2つの問題を一緒に解決します。ラボは、接続ごとにEOFで終わる1リクエストと、ASCIIバイトの大文字変換で、HTTPやTLSの実装ではありません。既存の行単位サーバーは変更せず、別のファイルに書きます。

ステップ

終了理由とソケットの所有権を分ける: Peer(sock, max_bytes=4096, expires_at=None)を作ってください。max_bytesはboolを除く1–65536のintです。expires_atがNoneならtime.monotonic()+30.0に決め、直接指定する場合は、boolを除く0以上の有限なint/floatでなければなりません。違反はValueErrorです。sockを非ブロッキングに設定し、sock・max_bytes・float(expires_at)を保持してください。inbox/outboxは空のbytes、read_eof/prepared/closedはFalse、reasonは空文字列です。finish(reason)は、最初の呼び出しでだけ、closed=True、reasonの保持、2つのバッファーを空にすることを行い、ソケットは閉じません。
EOFと上限超過を区別する: Peer.read_once()を実装してください。closedまたはread_eofなら、recvなしで0です。それ以外は、recv(min(4096, max_bytes+1-len(inbox)))を1回呼び出します。BlockingIOErrorは状態を維持して0、ほかのOSErrorはfinish('io_error')と0です。b''ならread_eof=Trueと0にするだけで、閉じません。受け取ったバイトはinboxに追加し、上限を超えたらfinish('request_limit')します。データがあったなら、受け取った長さを返します。expires_atは変更しません。
口を閉じたお客さんに返信を作る: Peer.prepare_reply()を作ってください。closed、EOFの前、prepared=Trueなら、何もしません。それ以外は、prepared=Trueと記録し、inbox.upper()をoutboxに移してからinboxを空にします。outboxが空のときにだけfinish('complete')します。何度呼び出しても、残った返信を再生成しません。文字列のデコードなしで、bytes.upper()を使います。
実際に出た分だけ消す: Peer.write_once()は、closedまたは空のoutboxなら、sendなしで0です。それ以外は、outbox[:4096]を1回sendし、返された長さだけ先頭から消してから、その長さを返します。BlockingIOErrorはバッファーを維持して0、ほかのOSError、またはsendの戻り値0は、finish('io_error')と0です。送信後にoutboxが空になり、read_eofならfinish('complete')します。
EOFのあとに読み取りの関心を切る: Peer.events()は、closedなら0です。それ以外は、read_eof=FalseのときEVENT_READ、outboxがあるときEVENT_WRITEをORしたintを返します。ソケットやバッファーは変更しません。EOFのあとの返信の準備は、ループがすぐに行うので、実際の登録マスク0は、終了した接続にだけ適用します。
1バイトで寿命が無限に延びないようにする: Peer.expire(now)は、closedでなく、now >= expires_atのときにだけfinish('deadline')します。nowは、呼び出し側が提供する有限の単調時刻です。EOFのあとに返信が残っていても同じように適用し、すでに終了した理由は変更しません。
登録を解除してからFDを回収する: モジュール関数retire(selector, peer)は、selector.unregister(peer.sock)のあと、peer.sock.close()を行います。まだ登録されていなくてKeyErrorが出ても、closeし、それ以外のエラーは、closeしたあとで再び伝えます。peer.reasonは変更しません。有効な開いたソケットが入力で、実際のcloseは、所有者であるループが1回だけ呼び出します。
沈黙したお客さんの隣で実際のTCP返信を完成させる: serve(peers)を作ってください。入力は、互いに異なる開いた接続ソケットを持つ、新しいPeerのlistで、関数がソケットの所有権を受け取ります。DefaultSelectorですべての接続を一緒に処理し、selectの待機は、0.05秒と、最も近い締め切りまでの時間のうち小さいほう(最小0)です。毎回の反復とI/Oの直前に、expire(time.monotonic())を確認します。READでread_onceのあとprepare_reply、WRITEでwrite_onceを呼び出し、関心マスクを更新します。終了した接続はretireで取り除き、例外が出ても、残りのソケットとselectorを回収します。すべて終了したら、入力順のreasonのlistを返し、空の入力は[]です。待ち受け・accept・スレッドの生成は、この関数では行いません。採点は、4つの実際のTCP接続と、前のステップの契約を再実行します。

参考

mkdir -p /root/halfcloseを実行してから、/root/halfclose/server.pyに保存してください。インストールや外部通信はありません。
python3 /opt/fixtures/halfclose/check.py 8 /root/halfclose/server.pyで、総合検査を実行します。8の代わりに現在のステップ番号を入れると、前のステップまで累積して検査します。
BlockingIOErrorは、まだ進めないという意味で、b''は受信の終了です。recv(0)を終了の検査に使わないでください。
完了は、ローカルの出力キューの消化であり、相手の業務処理の保証ではありません。採点は、実際に受信したバイトも別に確認します。
セッションが終了するとファイルが消えます。必要なコードは、終了する前に別に保管してください。

終了理由とソケットの所有権を分ける

Peer(sock, max_bytes=4096, expires_at=None)を作ってください。max_bytesはboolを除く1–65536のintです。expires_atがNoneならtime.monotonic()+30.0に決め、直接指定する場合は、boolを除く0以上の有限なint/floatでなければなりません。違反はValueErrorです。sockを非ブロッキングに設定し、sock・max_bytes・float(expires_at)を保持してください。inbox/outboxは空のbytes、read_eof/prepared/closedはFalse、reasonは空文字列です。finish(reason)は、最初の呼び出しでだけ、closed=True、reasonの保持、2つのバッファーを空にすることを行い、ソケットは閉じません。

状態の終了とFDの回収は別です。最初に残した失敗の理由を、あとの後始末の過程が上書きしないようにしてください。

EOFと上限超過を区別する

Peer.read_once()を実装してください。closedまたはread_eofなら、recvなしで0です。それ以外は、recv(min(4096, max_bytes+1-len(inbox)))を1回呼び出します。BlockingIOErrorは状態を維持して0、ほかのOSErrorはfinish('io_error')と0です。b''ならread_eof=Trueと0にするだけで、閉じません。受け取ったバイトはinboxに追加し、上限を超えたらfinish('request_limit')します。データがあったなら、受け取った長さを返します。expires_atは変更しません。

許容量にちょうど達したときも、次の1バイトかEOFを確認する必要があります。recv(0)は終了の証拠ではありません。

口を閉じたお客さんに返信を作る

Peer.prepare_reply()を作ってください。closed、EOFの前、prepared=Trueなら、何もしません。それ以外は、prepared=Trueと記録し、inbox.upper()をoutboxに移してからinboxを空にします。outboxが空のときにだけfinish('complete')します。何度呼び出しても、残った返信を再生成しません。文字列のデコードなしで、bytes.upper()を使います。

EOFのあとも、出力方向は生きています。preparedは、一部を送信したあとで、返信を最初から作り直すことを防ぎます。

実際に出た分だけ消す

Peer.write_once()は、closedまたは空のoutboxなら、sendなしで0です。それ以外は、outbox[:4096]を1回sendし、返された長さだけ先頭から消してから、その長さを返します。BlockingIOErrorはバッファーを維持して0、ほかのOSError、またはsendの戻り値0は、finish('io_error')と0です。送信後にoutboxが空になり、read_eofならfinish('complete')します。

正常な経路の短い応答だけで、部分送信を証明しないでください。EAGAINの次の呼び出しで、同じ残りのバイトが出ていく必要があります。

EOFのあとに読み取りの関心を切る

Peer.events()は、closedなら0です。それ以外は、read_eof=FalseのときEVENT_READ、outboxがあるときEVENT_WRITEをORしたintを返します。ソケットやバッファーは変更しません。EOFのあとの返信の準備は、ループがすぐに行うので、実際の登録マスク0は、終了した接続にだけ適用します。

送るものがあるという事実は、また受け取れるという意味ではありません。入力と出力の条件を別々に計算してください。

1バイトで寿命が無限に延びないようにする

Peer.expire(now)は、closedでなく、now >= expires_atのときにだけfinish('deadline')します。nowは、呼び出し側が提供する有限の単調時刻です。EOFのあとに返信が残っていても同じように適用し、すでに終了した理由は変更しません。

idleと総寿命を混ぜないでください。今回の契約の締め切りは、活動ではなく、生成の時点で固定されます。

登録を解除してからFDを回収する

モジュール関数retire(selector, peer)は、selector.unregister(peer.sock)のあと、peer.sock.close()を行います。まだ登録されていなくてKeyErrorが出ても、closeし、それ以外のエラーは、closeしたあとで再び伝えます。peer.reasonは変更しません。有効な開いたソケットが入力で、実際のcloseは、所有者であるループが1回だけ呼び出します。

finallyは、例外を隠す機能ではなく、所有するリソースを手放す場所です。KeyErrorとほかの例外を区別してください。

沈黙したお客さんの隣で実際のTCP返信を完成させる

serve(peers)を作ってください。入力は、互いに異なる開いた接続ソケットを持つ、新しいPeerのlistで、関数がソケットの所有権を受け取ります。DefaultSelectorですべての接続を一緒に処理し、selectの待機は、0.05秒と、最も近い締め切りまでの時間のうち小さいほう(最小0)です。毎回の反復とI/Oの直前に、expire(time.monotonic())を確認します。READでread_onceのあとprepare_reply、WRITEでwrite_onceを呼び出し、関心マスクを更新します。終了した接続はretireで取り除き、例外が出ても、残りのソケットとselectorを回収します。すべて終了したら、入力順のreasonのlistを返し、空の入力は[]です。待ち受け・accept・スレッドの生成は、この関数では行いません。採点は、4つの実際のTCP接続と、前のステップの契約を再実行します。

1人のお客さんの応答を待つ内部ループを作ると、また逐次サーバーになります。それぞれの準備完了イベントで、有限の入出力だけを行ってください。