Day 037 — Futureを手書きする。Wakerを保存し忘れるとタスクは永遠にハングする

📚 背景知識（読んでから問題へ）

Day 036では「async fnは呼び出しても即座には実行されない怠惰な値であり、誰か（ランタイム）がpollを呼ぶことで一歩ずつ前進する」というFutureのポーリングモデルを学びました。今日はその理解を一段深め、async fnというシンタックスシュガーを使わず、Futureトレイトを手書きで実装します。

なぜ手で実装する必要があるのか。実務でFutureを手書きすることは稀です（ほとんどの場合async fnで十分）。しかし、コールバックベースのC言語ライブラリやOSのI/O完了通知APIをasync/awaitで使える形にラップする場合、tokio・hyperなどのランタイム/ライブラリの内部実装を読む場合、独自のタイマー・独自のI/Oリアクターを実装する場合には避けて通れません。手書きすることで「.awaitの裏側で実際に何が起きているか」がコードとして目に見える形になります。

pollが「無駄に」何度も呼ばれてはいけない理由。Wakerの仕組みが正しく実装されていれば、あるFutureが完了するまでの間、pollは理論上「Pendingを返す1回」と「Readyを返す1回」の合計2回だけ呼ばれれば十分です。wake()を呼び忘れると再pollされずハングし、逆に的外れに呼びすぎると無意味なpollが連発するビジーポーリングが起きます。今日はこのpoll回数を実際にカウントして検証します。

📝 問題

以下の3つの要件を満たす、独自のTimerFutureを実装してください。

要件1（Future本体の実装）: 指定したDurationが経過するとPoll::Ready(())になるTimerFutureを、tokio::time::sleepを使わずに実装してください。

• 内部でstd::thread::spawnを使い、別スレッドでstd::thread::sleep(duration)した後に完了フラグを立てる
• Futureトレイトのpollメソッドを手書きで実装する（Pin<&mut Self>・Context・Pollを正しく使う）
• 完了前にpollされた場合は、cx.waker()をクローンして保存し、別スレッドが完了時に.wake()を呼べるようにする

要件2（poll回数の検証）: TimerFutureをラップし、pollが呼ばれた回数を数えるCountingFuture<F>を実装してください。このTimerFutureを.awaitした際、pollが正確に2回（1回目はPending、2回目はReady）しか呼ばれないことを、実際に実行して確認してください。

要件3（比較実験）: 同じ待ち時間（200ms）をtokio::time::sleep(...).awaitで実装した場合と、要件1のTimerFutureとで、実行結果（完了までのおおよその経過時間）が同等になることをmain関数内で両方実行し、経過時間をログに出して確認してください。

🔍 ヒント（段階的開示）

📋 コピーuse std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};

struct SharedState {
    completed: bool,
    waker: Option<Waker>,
}

struct TimerFuture {
    shared_state: Arc<Mutex<SharedState>>,
}

impl TimerFuture {
    fn new(duration: Duration) -> Self {
        // ここでスレッドをspawnし、sleep後にcompleted=trueとwake()する
        todo!()
    }
}

impl Future for TimerFuture {
    type Output = ();
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // ロック → completedを確認 → Ready or (waker保存してPending)
        todo!()
    }
}

✅ 模範解答

📋 コピー// Cargo.toml に以下を追加して検証すること:
// [dependencies]
// tokio = { version = "1", features = ["full"] }

use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};

/// TimerFuture と、完了を検知する別スレッドの間で共有する状態。
/// completed: タイマーが完了したかどうか
/// waker: pollが呼ばれるまではNone（まだ誰も待っていない）。
///        Pendingを返すたびに最新のWakerへ上書きしておく。
struct SharedState {
    completed: bool,
    waker: Option<Waker>,
}

/// 指定した時間が経過すると Ready(()) になる自前のFuture。
/// tokio::time::sleepを使わず、標準ライブラリのthread::sleepと
/// Waker手動管理だけで「ポーリングモデル」を再現する。
struct TimerFuture {
    shared_state: Arc<Mutex<SharedState>>,
}

impl TimerFuture {
    fn new(duration: Duration) -> Self {
        let shared_state = Arc::new(Mutex::new(SharedState {
            completed: false,
            waker: None,
        }));

        let thread_shared_state = Arc::clone(&shared_state);
        thread::spawn(move || {
            thread::sleep(duration);
            let mut state = thread_shared_state.lock().unwrap();
            state.completed = true;
            // take()で1度だけwake()する。既にpollされてWakerが
            // 登録されている場合のみ起こす必要がある。
            if let Some(waker) = state.waker.take() {
                waker.wake();
            }
        });

        TimerFuture { shared_state }
    }
}

impl Future for TimerFuture {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut state = self.shared_state.lock().unwrap();
        if state.completed {
            Poll::Ready(())
        } else {
            // pollが呼ばれるたびに最新のWakerへ更新する。
            // (異なるエグゼキュータ/タスクに移動する可能性があるため
            //  古いWakerを使い回すのは不正確)
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

/// 内側のFutureへpollを委譲しつつ、呼ばれた回数を数えるラッパー。
/// Waker管理が正しく実装されているかを検証する目的で使う。
struct CountingFuture<F> {
    inner: F,
    poll_count: Arc<AtomicUsize>,
}

impl<F: Future + Unpin> Future for CountingFuture<F> {
    type Output = F::Output;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.poll_count.fetch_add(1, Ordering::SeqCst);
        Pin::new(&mut self.inner).poll(cx)
    }
}

#[tokio::main]
async fn main() {
    println!("=== 要件1・2: 手書きTimerFutureとpoll回数の検証 ===");
    let start = Instant::now();
    let poll_count = Arc::new(AtomicUsize::new(0));
    let counted = CountingFuture {
        inner: TimerFuture::new(Duration::from_millis(200)),
        poll_count: Arc::clone(&poll_count),
    };
    println!("[{:>6?}] TimerFuture待機開始", start.elapsed());
    counted.await;
    let count = poll_count.load(Ordering::SeqCst);
    println!(
        "[{:>6?}] TimerFuture完了。poll()は{}回呼ばれた",
        start.elapsed(),
        count
    );
    assert_eq!(count, 2, "pollはPendingで1回・Readyで1回の合計2回だけ呼ばれるべき");

    println!();
    println!("=== 要件3: tokio::time::sleepとの経過時間比較 ===");
    let start2 = Instant::now();
    tokio::time::sleep(Duration::from_millis(200)).await;
    println!("[{:>6?}] tokio::time::sleep(200ms) 完了", start2.elapsed());

    let start3 = Instant::now();
    TimerFuture::new(Duration::from_millis(200)).await;
    println!("[{:>6?}] 手書きTimerFuture(200ms) 完了", start3.elapsed());
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn timer_future_completes_after_duration() {
        let start = Instant::now();
        TimerFuture::new(Duration::from_millis(50)).await;
        assert!(start.elapsed() >= Duration::from_millis(50));
    }

    #[tokio::test]
    async fn poll_is_called_exactly_twice() {
        let poll_count = Arc::new(AtomicUsize::new(0));
        let counted = CountingFuture {
            inner: TimerFuture::new(Duration::from_millis(30)),
            poll_count: Arc::clone(&poll_count),
        };
        counted.await;
        assert_eq!(poll_count.load(Ordering::SeqCst), 2);
    }
}

🪜 Step-by-Step 解説

💡 設計思想・なぜこう書くのか

Futureトレイトがpollという単一のメソッドだけを要求する最小限のインターフェースになっているのは、「どのように完了を通知するか」を実装者に委ねることで、あらゆる非同期処理（タイマー・ソケットI/O・チャンネル受信・GPU計算の完了通知など）を同じ抽象の上に統一的に表現できるようにするためです。Wakerという「起こす手段」だけを抽象化し、「何によって起こされるか」（今回は別スレッドのsleep、実際のI/Oではepoll/kqueue/IOCPなどのOSのイベント通知）を実装側の自由に委ねる設計は、Rustの「メカニズムとポリシーの分離」という思想の典型例です。

🛑 コンパイルエラーが出た場合

Wakerを保存せずにcx.waker()への参照だけを保持しようとすると、次のようなライフタイムエラーになります。

🌐 他言語との比較

C言語でのコールバックベースAPIをRustのasyncにラップする作業（「asyncラッパーを書く」）は、本問で書いたTimerFutureと全く同じ構造（SharedState＋Waker保存）を取ります。今日の実装は、実務でFFI境界を非同期化する際の直接的な雛形になります。

🏆 実務での使いどころ

• gRPC/HTTPクライアントライブラリの内部実装（hyper・tonicなど）は、OSのソケットI/OイベントをFuture::pollに橋渡しするために、本問と同様のWaker管理パターンを使っている
• レガシーなC言語ライブラリ（暗号化処理・ハードウェアドライバのSDKなど）をコールバックベースのままtokioエコシステムに統合する際、コールバックが呼ばれた瞬間に保存しておいたWakerをwake()する、というブリッジコードを書く必要がある
• 独自のタイムアウト機構やレート制限器（rate limiter）を作る場合、tokio::timeに頼らず本問のような手書きFutureでカスタムのスケジューリングロジックを組み込むことがある
• ライブラリ内部の挙動を調査するとき、本問のCountingFutureのようなpollカウンタを一時的に差し込んでデバッグする手法は実務でもそのまま使える

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: TimerFutureを複数同時にtokio::join!で待った場合、それぞれが独立したスレッドを生成してしまう非効率さに気づけるはずです。1本の管理スレッド（タイマーホイール）に統合して複数のタイマーをまとめて管理する設計に挑戦してみましょう
• 次回予告: tokioタスク・並行実行パターン — spawn・join!・select!の使い分けと、それぞれのキャンセル・エラー伝播の違いに進みます

🎯 自己評価

自分の回答

気づき・メモ