TT Lab
はじめる
学ぶ 学習パス コース

接続ひとつが遅くなり、残り全部が止まった

接続診断器：同時接続と安全なキャンセル

TT Labで続きを見る

目標

複数のTCP接続の試行を、制限されたバジェットで進め、拒否・締め切り・キャンセルを区別しながら、すべてのリソースを回収する診断器を作ります。

90分のラボです。基本のセッションは60分なので、期限が切れる前に＋時間で延長してください(最大180分)。セッションが終わるとファイルが消えます。重要な成果物は、終了前に別に保管してください。

なぜ重要なのか

接続の成功だけを見る診断器は、失敗の原因を隠し、キャンセルできないツールは、ユーザーに強制終了させることになります。今回のラボは、前の公平性・SO_ERROR・制御チャネルを、1つのループでつなげます。実際の通信は、ラボ環境のloopbackだけを使い、外部サーバーは検査しません。データの配信・HTTP・TLS・リトライ・DNS・サーバーのgraceful shutdownは、このラボの範囲ではありません。DefaultSelectorに基づく完了処理であり、ETの読み取りサーバーではありません。

検査は、ソースファイルをインポートして動作をテストします。ファイルの先頭でサーバーを起動したり、入力を待ったりしないでください。追加のパッケージや権限は必要ありません。/opt/fixtures/reactorの3つのprobe.pyは、別の概念観察用であり、修正したり提出したりする材料ファイルではありません。

ステップ

  1. 接続の試行の入力と初期状態を定義する: /root/reactor/client.pyでDial(sock, host, port, deadline=None)を実装してください。入力のsockは、呼び出し側が開いた、互いに異なるLinux AF_INET/SOCK_STREAMのソケットです。hostは数値のIPv4文字列だけ、portはboolを除く1–65535のintだけを受け付け、違反はValueErrorです。deadline=Noneならtime.monotonic()+30、指定する値は、boolを除く有限で0以上のint/floatでなければなりません。sockを非ブロッキングに変え、sock、endpoint=(正規化したhost, port)、floatのdeadline、state='new'、error=Noneを保持してください。生成に失敗したときのソケットの回収は、呼び出し側の責任です。

  2. 即時の成功と接続中の状態を区別する: /root/reactor/client.pyで、Dial.start()はnewのときにだけsock.connect_ex(endpoint)を1回呼び出します。0/EISCONNはconnected・error=0、EINPROGRESS/EALREADYはpending・error=None、それ以外のコードはfailed・error=コードです。OSErrorはerrnoを記録し、errnoがなければEIOです。newでなければ、呼び出しなしで既存のstateを返します。すべての経路でstateの文字列を返し、このLinux IPv4の契約では、EAGAINをpendingに含めません。

  3. 最初のSO_ERRORを保存する: /root/reactor/client.pyで、Dial.writable()はpendingのときにだけgetsockopt(SOL_SOCKET, SO_ERROR)を1回読み、0/EISCONNはconnected・0、それ以外のコードはfailed・該当コードとして記録します。OSErrorは、errnoまたはEIOで失敗として処理します。pendingでなければ照会しません。現在のstateを返します。呼び出し側は、WRITEの準備完了を観察したあとにだけ、pendingでこのメソッドを呼び出す必要があります。実際のTCPのlistenポートと、bindだけを行った拒否ポートも検査します。

  4. 待機キューでの締め切りとキャンセルの理由を保存する: /root/reactor/client.pyで、Dial.expire(now)は、new/pendingでnow >= deadlineのときにだけ、timed_out・ETIMEDOUTに変えます。nowは、呼び出し側が提供する有限の単調時刻です。Dial.cancel()は、new/pendingだけをcancelled・ECANCELEDに変えます。2つのメソッドは、すでに終了した状態とエラーを上書きせず、ソケットを直接閉じません。deadlineも延ばしません。

  5. 重複のない準備キューを作る: /root/reactor/client.pyで、ReadyQueue()にpush(item)、pop()、discard(item)、__len__を実装してください。itemは、Noneでなくハッシュ可能な接続の識別子です。pushは、すでにあれば無視し、popはFIFOで取り除いて返し、空ならNone、discardは該当する項目を取り除き、なくても安全でなければなりません。取り出した項目は再び追加でき、同じFDの別の世代の識別子は区別します。runでは、Dialオブジェクト自体をキーに使います。

  6. 停止状態を記録し、カーネルの待機を起こす: /root/reactor/client.pyで、Control()は、非ブロッキングのsocketpairのreader/writerと、最初はFalseのthreading.Event stop_eventを作ります。request_stop()は、Event.setのあと、writer.send(b'Q')を1回呼び出し、BlockingIOErrorだけを無視します。drain()は、reader.recv(4096)を1回呼び出してbytesを返しますが、BlockingIOErrorはb''、EOFはConnectionErrorです。close()は、readerを閉じるときに例外が出ても、writerまで閉じます。run終了後のrequest_stopの呼び出しは、契約の範囲外です。

  7. 上限とバジェットを守りながら、すべてのソケットを回収する: /root/reactor/client.pyで、run(dials, control, limit=8, budget=4, ready=None)を実装してください。dialsは、互いに異なる開いたソケットを持つ、互いに異なる新しいDialのlist(最大128個)、limitはboolを除く1–128のint、budgetはboolを除く1–64のintです。リストの型・個数・Dialの重複・newの状態と2つのバジェットを検証し、違反時はValueErrorで拒否し、このときの所有権は呼び出し側に残ります。DefaultSelectorの生成に成功したら、関数がすべてのデータソケットとControlを所有します。制御readerをREAD/data=Noneで登録し、readyがあれば1回呼び出します。毎ターン、停止を先に確認し、new/pendingに締め切りを適用します。停止なら、未完了をすべてcancelして終了します。1ターンに最大budget個を開始し、pendingを最大limit個、WRITE/data=Dialで登録します。すぐに終了した項目は回収します。待機は、最も近い締め切りまでの時間と1秒のうち小さいほう(最小0)で、準備キューが残っているか、待機リストに開始する余裕があれば0です。selectの結果の制御イベントはdrain、WRITEは重複のない準備キューに入れます。停止をもう一度確認してから、キューから最大budget個だけ取り出し、I/Oの直前に締め切りを確認してから、pendingのwritableを呼び出します。終了した項目は、キューから削除・unregisterしてからcloseします。例外が出ても、ほかのデータソケットとControl、selectorまで回収し、エラーを伝える必要があります。1つの回収の失敗が、残りの回収を妨げないようにしてください。空のリストでも、Controlを回収して[]を返します。正常終了は、入力順の{'state': 状態, 'error': 整数}のlistです。診断用の接続は、成功しても閉じ、再利用しません。

  8. 成功と失敗の原因を、損失なく集計する: /root/reactor/client.pyで、summarize(results)は、dictのlistだけを受け取り、違反はValueErrorです。各行は、stateがconnected/failed/timed_out/cancelledのいずれか、errorはboolを除くintでなければなりません。connectedは0、それ以外の状態は正の数で、timed_outはETIMEDOUT、cancelledはECANCELEDでなければなりません。pendingなどの未完了も、ValueErrorで拒否します。戻り値は、totalと4つの状態別の件数、errorsの辞書です。errorsは、connectedを除くerrnoの文字列をキーとして合算し、キーを並べ替えます。空の入力は、totalと4つの状態がすべて0、errors={}です。元のリストと行は変更しません。このステップは、前の実際のTCP・締め切り・キャンセル・バジェット・例外での回収も、再検査します。

参考

接続の試行の入力と初期状態を定義する

ipaddress.IPv4Addressで、名前解決を挟まずにアドレスを検証します。boolはintのサブタイプだという点に注意してください。

即時の成功と接続中の状態を区別する

書き込みの準備完了を待つあいだ、再接続を繰り返さないでください。1つのDialは、1回の接続の試行です。

最初のSO_ERRORを保存する

SO_ERRORは、読むと消えます。すでに確定した失敗を、次の照会の0で上書きしないでください。

待機キューでの締め切りとキャンセルの理由を保存する

待機キューにいた時間も、総寿命に含めます。状態の終了と、実際のFDの回収は、別の段階です。

重複のない準備キューを作る

dequeとsetの追加・削除を、一緒に管理してください。1つの作業が、重複した席で、ほかの接続の順番を奪ってはいけません。

停止状態を記録し、カーネルの待機を起こす

停止状態は先に記録し、通知はまとめられるようにしてください。通知のバイト1つを、個別のユーザーコマンド1つとして解釈しません。

上限とバジェットを守りながら、すべてのソケットを回収する

接続の開始バジェットと、完了処理のバジェットを、別々に数えます。完了イベントが一度に集中するテストで、2つの制限がどちらも守られる必要があります。

成功と失敗の原因を、損失なく集計する

0は成功の観察であり、空の情報ではありません。集計が終わった事実と、運用サービス全体の健全性は、区別してください。