C++面接：std::jthreadとstop_tokenを用いた協調的キャンセルの実装

プロンプトとコンテキスト

キューを消費するバックグラウンドワーカを実装します。シャットダウン時や上流のキャンセル時、メインスレッドは停止を要求します。開始済みのタスクは完了し、未開始の作業は保持されるか明示的に破棄されます。ビジーウェイト、データ競合、破棄中にスレッドが生存し続ける状態、および決して復帰しない条件変数の待機を回避してください。

C++20のstd::jthreadは、破棄時に停止を要求してjoinを行い、エントリ関数にstd::stop_tokenを注入できます。停止リクエストは協調的であり、任意のコードを強制終了することはできません。ワーカはポーリングを行うか、停止を認識するプリミティブを使用して待機する必要があります。

面接官が評価するポイント

停止トークンが共有状態に対するリクエストであり、非同期なスレッドの強制終了ではないことを理解しているか。
std::jthreadの自動joinを利用し、スレッドがアクセスしている間オブジェクトのメンバを生存させ続けているか。
condition_variable_anyのstop-tokenオーバーロードなどを用いて、ブロッキング待機を中断可能にしているか。
キュー、停止状態、タスクの例外、およびクリーンアップのライフタイムを安全に維持しているか。
空キューでの停止、停止の競合、タスクの例外、重複したrequest_stop、および破棄順序をテストしているか。

最初に明確にすべき質問

実行中のタスクは最後まで完了してよいか、またそのタスクは冪等であるか、または外部への副作用を持つか？
キューが閉じたとき、未消費のタスクは破棄されるか、転送されるか、それとも別のワーカによってドレイン（回収）されるか？
待機は中断可能か、それともキャンセルできないサードパーティのI/O呼び出しに依存しているか？
例外は記録されるか、joinするスレッドに伝播されるか、あるいはサービスの停止に使用されるか？
複数のスレッドが停止、破棄、または再起動を呼び出す可能性があるか？

30秒での回答

「std::jthreadでワーカを所有し、そのエントリ関数でstd::stop_tokenを受け取ります。ミューテックスでキューを保護し、停止対応のcondition_variable_anyを使用して待機します。起床後、停止、キューのクローズ状態、およびタスク状態を確認します。デキューされたタスクは定義されたキャンセルポイントまで実行され、クリーンアップを行います。破棄時は停止を要求してjoinし、決してdetachしません。共有状態はスレッドよりも長く生存します。テストでは停止の競合と例外をカバーします。」

ステップバイステップの解決策

ステップ1：キャンセルの規約を定義する

「停止要求」と「タスク完了」を分離します。停止要求は新しい作業を防止し、開始されたタスクは完了するか安全なポイントでキャンセル結果を返します。任意のサードパーティ呼び出しが即座にキャンセルされることを保証しないでください。

ステップ2：所有権を定義する

キュー、ミューテックス、条件変数、およびstd::jthreadを所有するオブジェクトを、スレッドよりも長く生存させます。破棄処理では、メンバを解放する前に停止を要求して待機します。破棄されたスコープへの参照をキャプチャしたり、遅延コールバックに生のthisを公開したりしないでください。

ステップ3：待機をキャンセル可能にする

condition_variable_anyのstop-token待機を使用するか、notify_allを呼び出すstop_callbackを登録します。述語はキューが空でないこと、クローズ状態、およびstop_requested()をチェックします。起床時にはロックを再取得して状態を再確認します。

cpp

std::jthread worker([this](std::stop_token st) {
  for (;;) {
    Task task;
    {
      std::unique_lock lock(mu_);
      cv_.wait(lock, st, [this, &st] {
        return closed_ || !queue_.empty() || st.stop_requested();
      });
      if (st.stop_requested() || (closed_ && queue_.empty())) return;
      task = std::move(queue_.front());
      queue_.pop_front();
    }
    run(task, st);
  }
});

ステップ4：タスクの停止ポイントと例外を処理する

タスクのフェーズ間でトークンをチェックし、部分的な書き込みを回避するために副作用の境界を定義します。スレッド境界で例外をキャッチし、タスクIDとエラーを記録して、ワーカを続行するか停止するかを決定します。エントリ関数から例外が外部へエスケープしないようにしてください。

ステップ5：正しい順序でクローズする

新しいタスクを拒否し、キューにクローズ済みのマークを付け、待機スレッドに通知し、停止を要求し、joinを行い、その後に初めてリソースを解放します。ドレインのタイムアウトと残余タスクの処理を定義します。request_stop()の重複呼び出しは安全であり、副作用を繰り返してはなりません。

ステップ6：競合をテストし動作を観察する

空キューでの停止、デキュー中の停止、同時停止呼び出し、破棄中の通知、タスクの例外、ブロッキングI/Oのタイムアウト、および繰り返しのクローズをテストします。停止レイテンシ、完了/キャンセルされたタスク、残存キュー、例外、およびjoin時間を記録します。競合検出のためにThreadSanitizerを実行します。

優れた回答例

「私はjthreadでワーカのライフタイムを管理し、stoptokenを受け取ります。ミューテックスがキューの状態を保護し、停止対応のconditionvariable_anyがクローズ状態、非空状態、および停止要求をチェックするため、停止によって待機から起床します。デキュー後にロックを解放します。タスクは安全なポイントでトークンをチェックし、トランザクションのクリーンアップを完了します。エントリ関数は例外をキャッチして記録します。」

「シャットダウンでは新しい作業を拒否し、closedを設定し、通知し、停止を要求してjoinします。決してdetachしたり、キューやロガーを早期に解放したりしません。テストでは、空キュー、デキューの競合、例外、重複停止、長時間タスクのタイムアウト、およびThreadSanitizerをカバーします。」

よくある間違い

stop_tokenを強制終了として扱う → リソースやトランザクションが破損する → 協調ポイントを定義する。
std::threadのjoinを忘れる → 異常終了またはダングリングスレッドが発生する → jthreadまたは明示的なライフタイム所有権を使用する。
1回の通知のみを待つ → 通知を逃すと永久にスリープする → 述語ループを使用し、停止時に起床する。
ロックを保持したままタスクを実行する → プロデューサやシャットダウンがブロックされる → デキュー後に解放する。
エントリ関数から例外をエスケープさせる → プロセスが異常終了する → スレッド境界でキャッチする。
スレッドを停止する前にメンバを解放する → Use-After-Free（解放後使用） → 停止、join、その後に状態を解放する。

フォローアップの質問と回答

jthreadのデストラクタは何を行いますか？

join可能な場合、破棄時に停止を要求してjoinします。タスクを強制終了することはありません。タスクが応答する必要があるため、joinは安全なポイントまで待機する場合があります。

停止リクエストによって条件変数を起床させることはできますか？

condition_variable_anyのstop-tokenオーバーロードは、停止が要求されたときに復帰します。カスタム待機では、通知を行うためのストップコールバックと、状態を再チェックする述語が必要です。

実行中のデータベース書き込みを即座にキャンセルできますか？

即座にできると仮定してはいけません。ロールバック可能または冪等なステップ、ドライバのタイムアウト/キャンセルサポート、およびコミット境界での停止チェックを使用してください。

停止の競合をどのように回避しますか？

クローズ状態、キュー、および停止を1つのライフサイクルプロトコルとして扱います。ロック下で遷移を実行し、状態変更後に通知を行います。停止とデキューが同時に発生するウィンドウに負荷をかけ、ThreadSanitizerを実行します。