Tokioは進捗を与えるが順序は与えない：100万タスクのスケジューリング

Tokioは進捗を与えるが順序は与えない：100万タスクのスケジューリング これは以前の記事「Your Rust Service Isn't Leaking — It Could Be the Allocator」の前編です。その記事では、いくつかのメモリallocatorが私たちのワークロード下でどのように異なる動作をしたかについて書きました。メモリの動作がallocatorと関連していることを突き止める前に、まずアプリケーション側からメモリ使用量を削減しようと試みました。 私たちのサービスはイベント駆動型でした： メッセージキュー（Kafka/Redis Streams/NATS）からイベントを読み取る 各イベントについて、処理するためのTokioタスクを生成する 私たちのタスクパターン 私たちのワークロードでは、各イベントは最大1000のユーザー・トークンを持っていました。各ユーザー・トークンについて、アウトバウンドコールを行い、I/Oを待ち、イベントに属するすべての応答を収集する必要がありました。以下は、ユーザー・トークンごとにTokioタスクをファンアウトし、応答をファンインして、応答イベントを生成するコードの簡略版です。 struct Event { payload: Bytes, // ~4KB user_tokens: Vec<String>, // 最大1000トークン // その他のフィールド } ... // メインループ内 { let event: Event = fetch_next_event().await; tokio::spawn(async move { let data = event.payload.clone(); let mut tasks = JoinSet::new(); for token in &event.user_tokens { let token = token.clone(); let data = data.clone(); tasks.spawn(async move { // アウトバウンドAPIにヒットし、応答を返す process(token, data).await }); } let mut responses = Vec::with_capacity(event.user_tokens.len()); while let Some(res) = tasks.join_next().await { responses.push(res) } generate_response_event(event, responses); }); } 私たちの主な焦点はスループットでした。上記のファンアウトは無制限ですが、実際には問題にならないだろうと想定していました。これらのTokioタスクは短命です。応答を受け取るとすぐに終了してドロップされ、通常は数ミリ秒以内に完了します。そのため、早く生成されたタスクも早く完了するだろうと想定していました。個々のアウトバウンドコールは順不同で完了する可能性がありましたが、新しいイベントがまだ入ってきている間でも、全体として早いイベントが先に完了することを期待していました。私たちの要件は、バーストが期待される時間内に完了するように、各アウトバウンドコールを可能な限り速くすることだけでした。 ログの外観 1000イベントのバースト中に、各イベントが約1000のユーザー・トークンを含み、合計約100万タスクが生成された際のログは以下のようになりました。 started: event 1, user 5 started: event 1, user 8 started: event 1, user 2 started: event 2, user 6 started: event 2, user 4 started: event 3, user 8 ... finished: event 779 finished: event 976 started: event 900, user 42 started: event 900, user 261 started: event 1, user 974 started: event 1, user 831 ... finished: event 5 finished: event 3 バーストは期待される時間内に完了しましたが、ログは、より早いイベントからのトークン・タスクがずっと後になって開始されていることを示していました。タスク間の厳密な順序は期待していませんでした。それらは任意の順序で完了してもよく、それは問題ありませんでした。驚いたのは、より早いタスクの一部が、最初のポーリングを受ける前に、その提出順序からどれだけ遅延したかということです。 Tokioのスケジューラ内部 Tokioのマルチスレッドランタイムには、固定数のワーカー・スレッド、各ワーカー用のローカルキュー、およびワーカー間で共有されるグローバルキューがあります。各ワーカーは容量256タスクのローカルキューを持ち、オーバーフローするとタスクの半分をグローバルキューに移動します。ワーカーはまず自身のローカルキューからプルすることを好み、時々グローバルキューをチェックし、アイドル時には他のワーカーから盗みます。 タスクが独立してスケジューリング可能になると、Tokioはそれらがどのイベントによって作成されたかを知らなくなります。それらは実行可能なタスクとなり、ポーリングされるために競合します。単純化された図は次のようになります。 global queue +------------------------------+ | * overflow from local queues | | * remotely scheduled tasks | | * mixed older/newer work | +--------------+---------------+ | v +-------------------------+-------------------------+ | | | v v v worker 0 worker 1 worker 2 +-------------+ +-------------+ +-------------+ | local queue | | local queue | | local queue | | up to 256 | | up to 256 | | up to 256 | +------+------+ +------+------+ +------+------+ | v poll task このアーキテクチャでは、イベントが1000のユーザー・トークン・タスクにファンアウトすると、それらのタスクは他のイベントからのトークン・タスク、JoinSetで待機している親イベント・タスク、およびI/O準備完了からウェイクアップしたタスクと混ざり合います。これほど多くのタスクが同時に提出されると、Tokioは早いタスクが先にポーリングされることを保証しません。異なるイベントからのタスクは利用可能なワーカーキューを競合し、キューのオーバーフローやワーク・スティーリングのようなスケジューリング決定は、タスクがピックアップされる順序を変更する可能性があります。その結果、より早いイベントからのいくつかのタスクは、ずっと後になって最初のポーリングを受け取ることになりました。 コアな違いは次のとおりです：タスク作成 != タスクポーリング != タスク完了 Tokioは各タスクの進捗を継続的に行いますが、メモリは同時にライブなタスクの数に影響されます。各Tokioタスクはいくらかの状態を持ち、その状態は個々には大きくなくても、数千ものタスクが同時にライブになると、ピークメモリ使用量に加算されます。そして、私たちのワークロードにおけるより早いイベントからのいくつかのタスクはバーストの最後まで生き残ったため、それらの親イベント・タスクも生き残ったままになり、イベントの状態をメモリ内に保持し、ピークメモリ使用量を増加させました。 タスク生成には上限が必要 Tokioは、タスク数が制限されており、かつタスクがワーカー・スレッドをブロックしないと仮定した場合、公平なスケジューリングを保証します。私たちのコードには、当初、タスク生成に上限がありませんでした。 イベントを可能な限り速く読み取る └── イベントごとに1つのイベントタスクを生成する └── イベントごとに最大1000のトークンタスクを生成する Tokioは与えられたタスクをすべて受け取り、それらの進捗を継続します。Tokioの公平性保証で想定されている上限は、アプリケーション側から提供される必要があります。これらの上限はワークロードによって異なります。私たちのケースでは、イベントレベルの公平性を求めていました。単一イベントに属するすべてのトークン・タスクが、その提出時間に近くポーリングされ、完了することを望んでいました。 これを達成するために、Semaphoreを使用して一度に処理できるイベントの数を制限しました。 一度にN個のイベントのみを処理に許可する └── イベントごとに1つのイベントタスクを生成する └── イベントごとに最大1000のトークンタスクを生成する Tokioが一度に見るタスクの最大数（公平性に影響する）は、各タスクがポーリングで返るのにかかる時間にも依存します。Semaphoreのカウントを見つけるために、試行錯誤を行いました。この制限された構造があっても、バーストは期待される時間内に完了し、ピークメモリを大幅に削減したという重要な詳細があります。一般的に、Semaphoreを追加するとサービスの全体のスループットに影響を与えるように見えるかもしれませんが、私たちにとってはそうではありませんでした。 Tokioの問題ではない これはTokioの問題ではありませんでした。100万のライブタスクがあっても、Tokioはバーストが時間通りに完了するまで進捗を続けました。違いは、私の想定（イベント全体が早く開始されたら早く完了すべき）と、Tokioランタイムが見ているもの（タスクはタスクであり、どのイベントから来たかは関係ない）との間にありました。これは、正しい性の問題がなかったため、私が見落としやすい点でした。途中でドロップされたタスクやサービスに顕著な遅延はなく、スループット要件も満たされていました。単に、アプリケーションロジックのどこにも現れない理由で、予想よりも高いメモリのスパイクとして現れただけでした。 結論 早く生成されたからといって、早くポーリングされるとは限りません。 早くポーリングされたからといって、早く完了するとは限りません。 Tokioに尊重してほしいアプリケーションレベルの公平性の単位（私たちのケースではイベント）がある場合、ランタイムはその存在を知りません。 上限はアプリケーション側で追加する必要があります。 そして、Tokioタスクを生成する前に、一度にライブなタスクの最大数がいくつになる可能性があるかを問い、それがメモリにどのように影響するかを確認してください。場合によっては、タスクが生きている間に保持している他のリソースにも影響を与える可能性があります。