問題と背景
この質問は、C++26 std::execution におけるコアオブジェクトモデルを評価します。最小限の非同期操作から始めて、sender の遅延記述、receiver の完了規約、connect によって生成される operation state、そして start が計算を開始するタイミングについて説明してください。
面接官が見ているポイント
重要な違いは、遅延した sender の記述と、接続後に作成される operation state との間にあります。connect が状態を構築し、start が処理を開始します。set_value、set_error、set_stopped がパイプライン内をどのように伝播するか、スケジューラの変更がスレッドアフィニティに影響するか、そしてキャンセルによって新たな副作用がどのように防がれるかを説明してください。
最初に確認すべき明確化のための質問
ワークロードとバックプレッシャー
バッチサイズ、許容される並行性、入力の順序、失敗した書き込みのリトライポリシーについて質問してください。その回答によって、sender の並行性を制限するか、キューに上限を設けるかが決まります。
スケジューリングと停止ソース（stop source）
I/O スケジューラと CPU スケジューラがどのように公開されているか、停止がタイムアウトによるものかユーザー操作によるものか、そして既に発行されたシステムコールがどのように完了するかを尋ねてください。停止要求はスレッドの強制終了ではありません。
リソースとコミットセマンティクス
ファイルハンドル、バッファ、一時ファイルの所有権、書き込みが冪等であるかどうか、部分的なバッチがロールバック可能かどうかを明確にしてください。これにより、set_error 後のクリーンアップ処理が定義されます。
30秒の回答フレームワーク
「読み取り、パース、書き込みを sender アダプタで記述し、スケジューラアダプタを使用して実行コンテキストを切り替えます。パイプラインは遅延状態を維持します。connect が operation state を作成し、start がそれを開始します。各ステージは成功値を下流へ、エラーを seterror 経由で、キャンセルを setstopped 経由で送信します。1 つの stop token がすべてのステージに届きます。停止パスは副作用の前にチェックを行い、所有者は発行済みの I/O を完了・クローズし、一時ファイルをクリーンアップします。」
詳細な解決手順
ステップ 1: 値とエラーの境界を定義する
各ステージの入力型と出力型を定義し、回復可能な失敗を明示的なエラー sender に変換します。例外がスケジューラの境界を越えないようにし、最終的な receiver に成功、失敗、または停止を記録させます。
ステップ 2: 遅延パイプラインを構成する
let_value、then、または同等のアダプタを使用して、読み取り、パース、書き込みを接続します。構成（Composition）は、スレッドの割り当てや I/O の実行を行わずに記述を構築します。operation state は、コールバックより長く存続する必要がある共有状態を所有します。
ステップ 3: 接続して開始する（connect & start）
connect(sender, receiver) を呼び出して operation state を取得し、それを有効なスコープ内に維持した上で start を呼び出します。receiver は operation state よりも長く存続しなければなりません。非同期コールバックが破棄されたスタックオブジェクトを参照することはできません。
ステップ 4: 実行コンテキスト間を移動する
I/O が完了した後、スケジューラ sender を使用してパース処理を CPU プールに移動し、その後書き込みをバウンドされた I/O プールに戻します。キューの容量と公平性を記録してください。ブロッキングを伴う書き込みを無制限の汎用プールに配置しないでください。
ステップ 5: 停止とバックプレッシャーを伝播する
中断可能なすべてのステージに stop token を渡します。停止後は新しいバッチをエンキューしないでください。処理中のシステムコールは、その API に従ってキャンセルまたは完了させます。キューが一杯の場合、スロットリング sender が上流の処理を一時停止してメモリを制限します。
ステップ 6: エラーと部分的な副作用を処理する
一時ファイルに書き込むか、バッチシーケンスを記録してからアトミックにコミットします。set_error は下流のクリーンアップとハンドルのクローズをトリガーします。リトライによって書き込みが重複しないよう、制限と冪等性キー（idempotency key）が必要です。
ステップ 7: 並行性と生存期間を検証する
複数のスケジューラ、停止の競合、パースの失敗、部分的な書き込み、および receiver の早期破棄をテストします。スレッドアナライザを使用してデータ競合を検出し、キュー待機時間、スループット、停止レイテンシ、および解放されていない operation state を測定します。
高品質な回答例
読み取り sender はバッチを発行し、パース sender は CPU スケジューラ上で実行され、書き込み sender はバウンドされた I/O スケジューラ上で実行されます。パイプラインは依存関係を記述するだけであり、connect が operation state を作成し、start がそれを起動します。すべてのステージが値、エラー、停止の完了を処理し、1 つの stop token を共有します。停止は新しいバッチをブロックし、発行済みの I/O を安全にクローズさせ、一時ファイルとバッチ ID を使用して冪等なリトライを実現します。テストでは、スケジューラのホップ、バックプレッシャー、停止の競合、および receiver の生存期間を網羅します。
よくある間違い
- 間違い: 構築された sender が既に実行中であると想定すること。 → 理由: sender は遅延評価されます。 → 対策: connect/start の境界を明示してください。
- 間違い: 例外は処理するが、stopped 完了を処理しないこと。 → 理由: 停止は独立した完了チャネルです。 → 対策: seterror と setstopped の両方を実装してください。
- 間違い: キャンセル時にスレッドを強制終了すること。 → 理由: ファイルハンドルや部分的な書き込みを所有している可能性があります。 → 対策: stop token を伝播させ、安全にアンワインドしてください。
- 間違い: 上限なしでバッチをエンキューすること。 → 理由: バックプレッシャーがないとメモリが枯渇する可能性があります。 → 対策: 並行性、キュー容量、リトライ回数に上限を設けてください。
フォローアップの質問と回答
フォローアップ 1: なぜ future を直接使用しないのですか？
sender/receiver は、スケジューリング、キャンセル、および 3 つの完了チャネルを合成可能にし、接続時の生存期間制御を提供します。通常、future では停止やエラーの伝播に追加の規約が必要になります。
フォローアップ 2: start の後に sender を破棄できますか？
一時的な sender の記述は破棄できますが、operation state、receiver、およびキャプチャされたリソースは完了するまで有効でなければなりません。タスクオブジェクトまたはスコープがそれらを所有する必要があります。
フォローアップ 3: set_stopped はすべての副作用をロールバックしますか？
いいえ。これは停止完了を報告するだけであり、既に発行された I/O はロールバックされない場合があります。整合性を保つためには、一時ファイル、冪等なコミット、または補償処理が必要です。
フォローアップ 4: 書き込みが重複しないことをどのように証明しますか？
一意で安定したバッチ ID を割り当て、書き込み前にコミット済み ID を確認し、リトライ時には欠落している ID のみを書き込むようにします。障害、再起動、停止の競合を注入し、コミットログと最終ファイルを比較します。