[Rust] osaka.rs メモ


[今週のThis Week in Rust](https://this-week-in-rust.org/blog/2019/02/05/this-week-in-rust-272/)に出てきた[osaka.rs](https://github.com/aep/osaka)にちょっと興味惹かれたので読んでみたときの個人メモ

# osaka.rsとは？

[mio](https://github.com/carllerche/mio)をベースに、`async`もどきの`#[osaka]`と`await!`もどきの`sync!`を実装したお試し非同期フレームワークという感じ。

# `#[osaka]`

`#[proc_macro_attribute]`については

https://doc.rust-lang.org/reference/procedural-macros.html#attribute-macros

```rust:osaka-macro/lib.rs
    let output = match f.decl.output.clone() {
        syn::ReturnType::Default => quote! {()},
        syn::ReturnType::Type(_, t) => quote! {#t},
    };

    f.decl.output = match syn::parse((quote! {
        -> osaka::Task<#output>
    }).into(),
    ) {
        Ok(v) => v,
        Err(e) => return e.to_compile_error().into(),
    };
```

返り値を`osaka::Task<>`でラップしている。

`syn::ReturnType::Default`は返り値無しの関数の場合。

```rust:osaka/lib.rs
    let oblock = f.block;
    f.block = match syn::parse(
        (quote! {{
            use std::ops::Generator;
            let mut l = move||{
                #oblock
            };

            let a = match unsafe { l.resume() } {
                std::ops::GeneratorState::Complete(y) => {
                    return osaka::Task::immediate(y);
                }
                std::ops::GeneratorState::Yielded(y) => {
                    y
                }
            };
            osaka::Task::new(Box::new(l),a)
        }})
        .into(),
    ) {
        Ok(v) => v,
        Err(e) => return e.to_compile_error().into(),
    };
```

関数のブロックをラッピングして[`std::ops::Generator`](https://doc.rust-lang.org/std/ops/trait.Generator.html)を作ってすぐに`resume()`で実行し、作ったジェネレータと中断結果(完了した場合は`osaka::Task::immediate`)を返す関数を作り、それを`f.block`に代入することで元の関数を作り変えている。

この`#[osaka]`はどんな関数にも付けられるというものではなく、ブロック内部で`yield`(または後述の`sync!`)を呼んでいる必要があるため、どちらかというとジェネレータを普通の関数に見せかける仕組みという方が正しいように思う。

# `sync!`

```rust:osaka/lib.rs
macro_rules! sync {
    ($task:expr) => {{
        use osaka::FutureResult;
        use osaka::Future;
        loop {
            match $task.poll() {
                FutureResult::Done(y) => {
                    break y;
                }
                FutureResult::Again(y) => {
                    yield y;
                }
            }
        }
    };};
}
```

引数の`$task`は`#[osaka]`付きの関数(?)で生成された`osaka::Task`またはジェネレータ。
`yield`を上へと伝搬させていく`await!`もどきの役割。

# `Task::run()`

ブロックしてイベントを待つ。

```rust:osaka/lib.rs
                    a.poll.poll(&mut events, timeout).expect("poll");

                    for token in &a.tokens {
                        token.active.store(0, Ordering::SeqCst);
                    }
                    for event in &events {
                        for token in &a.tokens {
                            if event.token() == token.m {
                                debug!("token {:?} activated", token.m);
                                token.active.store(1, Ordering::SeqCst);
                            }
                        }
                    }
```

`mio::Poll::poll()`でイベントを待って、来たら対応する`token.active`を有効にする。

```rust:osaka/lib.rs
            if let FutureResult::Done(v) = self.poll() {
                return v;
            }
```

`poll()`を呼んでイベントを処理。`Done`なら上に返る。`Again`なら`loop`で再度`mio::Poll::poll()`のイベント待ちへ。

# `Task::poll()`

`osaka::Future::poll()`の実装。


```rust:osaka/lib.rs
        match self {
            Task::Immediate{r} =>  {
                return FutureResult::Done(r.take().expect("immediate polled after completion"));
            }
            Task::Later{f,a} =>  {
                let mut ready = false;

                if let Some(deadline) = a.deadline {
                    if Instant::now() >= deadline {
                        debug!("task wakeup caused by deadline");
                        a.deadline = None;
                        ready = true;
                    }
                }

                if !ready {
                    for token in &a.tokens {
                        if token.active.load(Ordering::SeqCst) > 0 {
                            debug!("task wakeup caused by token readyness");
                            ready = true;
                            break;
                        }
                    }
                }

                if ready {
                    match f.poll() {
                        FutureResult::Done(y) => {
                            return FutureResult::Done(y);
                        },
                        FutureResult::Again(a2) => {
                            *a = a2;
                        }
                    }
                }

                FutureResult::Again(a.clone())
            }
        }

```

デッドラインに到達、もしくはイベントが来ていれば、`osaka::Future`(ジェネレータ or `osaka::Task`)に対して`Future::poll()`を実行し`FutureResult`を返す。

動かすものがないならそのまま`Again`を返す。

# `Again`

```rust:osaka/lib.rs
#[derive(Clone)]
pub struct Again {
    tokens:     Vec<Token>,
    deadline:   Option<Instant>,
    poll:       Arc<mio::Poll>,
}
```

イベント待ちをするための情報。`Task::Later`のメンバーとして`sync!`で伝搬していき、`Task::run()`でイベント待ちするときに利用される。

# README.mdの例

以上を踏まえてREADME.mdを見てみる

## 概要

`resolve()`では引数の`poll`に待つイベントを登録し、イベントがあるごとに`yield poll.again`から再開して応答を受信。
`test()`では`resolve()`で作ったタスクを`sync!`で同期的に待つだけ。

## 詳細

```rust:README.md
#[osaka]
pub fn resolve(poll: Poll, names: Vec<String>) -> Result<Vec<String>, Error> {
    //...
    let sock = UdpSocket::bind(&"0.0.0.0:0".parse().unwrap()).map_err(|e| Error::Io(e))?;
    let token = poll
        .register(&sock, mio::Ready::readable(), mio::PollOpt::level())
        .unwrap();
    //...

    // wait for a packet
    let pkt = match loop {
        // wait for the token to be ready, or timeout
        yield poll.again(token.clone(), Some(Duration::from_secs(5)));
        if now.elapsed() >= Duration::from_secs(5) {
            // timeout
            break None;
        }
        // now the socket _should_ be ready
        let (len, from) = match sock.recv_from(&mut buf) {
            Ok(v) => v,
            Err(e) => {
                // but just in case it isn't lets re-loop
                if e.kind() == std::io::ErrorKind::WouldBlock {
                    continue;
                }
                return Err(Error::Io(e));
            }
        };
    }

    // do stuff with the pkt
    // ...
}

pub fn test(poll: Poll) -> Result<(), Error> {
    let mut a = resolve(
        poll.clone(),
        vec![
            "3.carrier.devguard.io".into(),
        ],
    );
    let y = osaka::sync!(a);
    println!("resolved: {:?}", y);
    Ok(())
}

pub fn main() {
    tinylogger::init().ok();
    let poll = osaka::Poll::new();
    test(poll).run().unwrap();
}
```

以下雑多なメモ

これ、`test()`に`#[osaka]`付ける必要あるのではないだろうか。以降`test()`に`#[osaka]`付いてる前提で話を進める。
あとこのままだといろいろコンパイルは通らないので動かすためには結構修正必要。

`#[osaka]`の効果で`resolve()`では`osaka::Task<>`が返る。
最初の実行では引数の`poll`に対するイベント登録をした後、`yield poll.again();`で
`osaka::Task::Later{Box<Generator>, Again}`を返す。

`test()`も`osaka::Task<>`を返す。`test()`のブロックもジェネレータ化されているので、最初の呼出で`resolve()`を呼んで`sync!`の中で上の`Later`に対して`poll()`を実行する。
最初の呼び出しては通常イベントは発生していないので上に`Again`が返って、その`Again`を元に`Task`が作られて`test(poll)`の返り値となる。

上の返り値にさらに`Task::run()`を実行する。`Task::run()`はブロッキングメソッドで`mio::Poll::poll()`を使って実際にブロッキングでイベント待ちをおこなう。イベントが来た場合は`Task::poll()`を呼び出す。あとは「`mio::Poll::poll()`して`Task::poll()`」の処理を全ての処理が終わるまで繰り返す。