Rust の Future について

use std::time::Duration;

async fn request_to_service1() -> u64 {
    tokio::time::sleep(Duration::from_secs(3)).await;
    3
}

async fn request_to_service2() -> u64 {
    tokio::time::sleep(Duration::from_secs(5)).await;
    5
}

// これだと service1, service2 へのリクエストが順に行われるので 3 + 5 秒かかる
pub async fn sample1() {
    let res1 = request_to_service1().await;
    let res2 = request_to_service2().await;
    println!("{}, {}", res1, res2);
}

// これだと 5 秒で済む
// 先に task を spawn しておく必要あり
// (実際は直接 Future 渡しても問題無さそうだが join の実装次第になってしまうはず)
pub async fn sample2() {
    let h1 = tokio::task::spawn(request_to_service1());
    let h2 = tokio::task::spawn(request_to_service2());
    let res = futures::future::join(h1, h2).await;
    println!("{:?}", res);
}

use pin_project::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

#[pin_project]
struct MyJoin<T: Future, U: Future> {
    #[pin]
    fut1: T,
    #[pin]
    fut2: U,
}


// 2 つの Future を待ち受ける Future
// あくまで擬似的な実装であることに注意。
// Future の poll は Ready を返したあと再度呼ばれた時の挙動が未定義なので実際は実行時に panic する。
impl<T: Future, U: Future> MyJoin<T, U> {
    pub fn new(fut1: T, fut2: U) -> MyJoin<T, U> {
        MyJoin { fut1, fut2 }
    }
}

impl<T: Future, U: Future> Future for MyJoin<T, U> {
    type Output = (T::Output, U::Output);

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let futures = self.project();
        let res1: Poll<T::Output> = futures.fut1.poll(cx);
        let res2: Poll<U::Output> = futures.fut2.poll(cx);
        match (res1, res2) {
            (Poll::Ready(res1), Poll::Ready(res2)) => Poll::Ready((res1, res2)),
            _ => Poll::Pending,
        }
    }
}

pub async fn sample() {
    let h1 = tokio::task::spawn(async {
        tokio::time::sleep(Duration::from_secs(3)).await;
        println!("fut1");
    });

    let h2 = tokio::task::spawn(async {
        tokio::time::sleep(Duration::from_secs(5)).await;
        println!("fut2");
    });

    MyJoin::new(h1, h2).await;
}

use futures::FutureExt;
use std::time::Duration;

async fn f(i: u64) -> u64 {
    tokio::time::sleep(Duration::from_secs(i)).await;
    println!("finish {}", i);
    i
}

pub async fn sample() {
    // f(3) がはじめに完了するはず
    let h1 = tokio::spawn(f(5));
    let h2 = tokio::spawn(f(4));
    let h3 = tokio::spawn(f(3));
    let (res, remaining) = futures::future::select_ok(vec![h1, h2, h3]).await.unwrap();
    println!("{:?}", res);

    // f(5), f(4) については h1, h2 を drop すればそれ以降実行は進まない
    // あるいは明示的に abort する
    // (とはいえタスク目線だと abort されていることに気付くのは await しているタイミングっぽいので
    // この例だと結局最後まで実行される)
    for h in remaining {
        h.abort();
    }
}

use pin_project::pin_project;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

#[pin_project]
struct MySelect<T: Future> {
    #[pin]
    fut1: T,
    #[pin]
    fut2: T,
}

impl<T: Future> MySelect<T> {
    pub fn new(fut1: T, fut2: T) -> MySelect<T> {
        MySelect { fut1, fut2 }
    }
}

impl<T: Future> Future for MySelect<T> {
    type Output = T::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let futures = self.project();
        if let Poll::Ready(res) = futures.fut1.poll(cx) {
            return Poll::Ready(res);
        }
        if let Poll::Ready(res) = futures.fut2.poll(cx) {
            return Poll::Ready(res);
        }
        Poll::Pending
    }
}

pub async fn sample() {
    let h1 = tokio::task::spawn(async {
        tokio::time::sleep(Duration::from_secs(3)).await;
        println!("fut1");
    });

    let h2 = tokio::task::spawn(async {
        tokio::time::sleep(Duration::from_secs(5)).await;
        println!("fut2");
    });

    // should print "fut1"
    MySelect::new(h1, h2).await;
}

async fn factorial(x: u32) -> u32 {
    return if x == 0 {
        1
    } else {
        x * factorial(x - 1).await
    };
}

// これだとサイズが決まらないので、
struct Foo {
    foo: Box<Foo>,
}


// 例えばこうする必要がある
struct Foo {
    foo: Box<Foo>,
}

use futures::future::BoxFuture;
use futures::FutureExt;

// in futures-core
// pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

fn factorial(x: u32) -> BoxFuture<'static, u32> {
    async move {
        return if x == 0 {
            1
        } else {
            x * factorial(x - 1).await
        };
    }
    .boxed()
}

// in alloc::boxed

#[stable(feature = "futures_api", since = "1.36.0")]
impl<F: ?Sized + Future + Unpin, A: Allocator> Future for Box<F, A>
where
    A: 'static,
{
    type Output = F::Output;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        F::poll(Pin::new(&mut *self), cx)
    }
}

// in std::future::Future

#[stable(feature = "futures_api", since = "1.36.0")]
impl<P> Future for Pin<P>
where
    P: ops::DerefMut<Target: Future>,
{
    type Output = <<P as ops::Deref>::Target as Future>::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        <P::Target as Future>::poll(self.as_deref_mut(), cx)
    }
}

use std::time::Duration;

// 一秒に一回処理を行う
// 処理の間は必ず 1 秒空けられる
async fn f() {
    loop {
        tokio::time::sleep(Duration::from_secs(1)).await;

        // do what we want here
        println!("do f");
    }
}

// 一秒に一回処理を行う
// 処理に 1 秒以上の時間がかかった場合はすぐに処理が行われる
async fn g() {
    let mut interval = tokio::time::interval(Duration::from_secs(1));
    loop {
        interval.tick().await;

        // do what we want here
        println!("do g");
    }
}