2019-09-29に更新

Rustの並列処理を学ぼうぜ☆（＾～＾）？＜その３＞

Rust 並列処理

読了目安：19分

Tokioクレートを学ぼうぜ☆？

「　あと２か月後の　2019年11月7日（金）に　Rust 1.29　が出てきて Rustに Async/Await　が言語実装されたら
これからTokioクレートを覚える意味がない上に　悪い癖が付くと思うが、
２年間隔で言語仕様を更新しますと言った Java 1.7 は６年待たされたからな☆　延期はありうる☆」

「　公式なんか信じてはいけないぜ☆」

「　Tokioクレートを覚えるのではなく、非同期処理を覚えなさい」

Crate tokio
Hello World!
Rust と非同期 IO の歴史
 できる！futures入門

「　2006年に開発が始まったとされる Rust 言語の非同期処理は　なぜ 2019年11月まで遅れているのかというと
2016年に AlphaZero が人間の棋譜を使わずに機械学習でイ・セドルに４勝１敗　するまで
ディープ・ラーニングに　使い道があると思われてなかったからだと思うが☆、」

「　ディープ・ラーニングの　精度　がビジネスに効く今、
精度を上げるため　の　並列処理、マルチ・コア・プログラミング　を今どき　書けなければ、
３０年後の例えば　２０５０年に　使われているプログラミング言語　に入らないから
尻に火が付いて　せっせと　慌てているのだろう☆」

「　お父んは　なんで Rust なんか使ってるんだぜ☆？　安牌を捨てるなら　Python ＋ Go　なのに……☆」

「　平岡さんが Rust を使うとか耳にして　流行り　に飛びついただけなのよ！
飛びついたはいいものの、　流行ってなかった　のよ！」

「　そういえば　お父んは　Java 1.1 とか、 Paradox とか触ってたらしいな☆」

「　先輩は　Delphi　を使っていた☆　もう耳にもしない☆」

「　Java Script とか、 C# とか、 C++ とか、お前らがよく耳にするプログラム言語は
Ｗｅｂのクライアント・サイド書くなら Java Script とか、
Unityでマルチ・プラット・フォーム対応のゲーム書くなら C# とか、
これをするなら　この言語、と言ったような　利用シーンを持っていて　シェアを獲っている☆」

「　だいたい　プログラム言語が生き残るルートは　２つあるようだぜ☆
大きいターゲットのシェアを獲るか、くそレガシーとして生き残る　かの２つだぜ☆」

「　レガシーは生き残らないのよ、まだ死に損なっているだけなの」

「　Tokioクレートも　十分レガシーだろう……☆
なんか　レガシーばっか　やることになるな……☆」

非同期の例外処理

「　非同期処理を覚えていこう☆　よく目にするのは　非同期の例外処理　だぜ☆」

「　非同期の例外処理って何だぜ☆？　同期の例外処理と何が違うんだぜ☆？」

「　まず　例外処理とは何か、を説明する☆」

「　りんご　が袋に入っているとしよう☆
奴隷には　りんごを３個～５個　袋に詰めなさい、と指示しておいた☆」

「　奴隷は労働力として大切にされていたらしいな☆　使いつぶしても代わりの利く日本人とは違って☆」

「　労働力じゃなければ大切にされなかったんだけどね」

「　３個～５個じゃない袋も　混ざっているだろう☆
もしかすると　梨　が混ざってるかもしれないしな☆　そういうものは　エラー　として個別の対応をする☆
りんごを袋から取り出して　詰める工程に差し戻して　やりなおしたり、
めんどくさければ　捨ててもいい☆」

「　それとは別に☆」

「　りんご　は　いっぱいあるのに、袋が　０枚　だったら　奴隷たちはどうするだろうか☆？」

「　仕事を休んでビールでも飲むぜ☆」

「　永遠にエラーにするのも　無駄よねぇ」

「　そういうときは　例外処理　として　主に報告を上げ、
安全な状態にして　作業をストップすることを目指す☆
あるいは　リトライ　を目指す場合もあるが、上図の状況では　リトライしても無駄なんで　終われだぜ☆」

Enum std::result::Result

「　↑Rust言語では　エラー処理　の書き方の模範があるので　まず　それを説明しよう☆」

「　std::result::Result　という列挙型がある☆
処理がグッドな終わり方をしたら　Ok　と一緒に値を返し、
処理がバッドな終わり方をしたら　Err　と一緒に値を返せだぜ☆」

let server = listener.incoming().for_each(|socket| {
        Ok(())
    })
    .map_err(|err| {
        println!("Error = {:?}", err);
    });

「　↑例えばこういうコードを目にすることがあるが……☆」

let server = listener.incoming().for_each( ☆ ).map_err( ☆ );

「　コールバック関数を省略すると　↑こういう見た目をしている☆」

「　Ｃ言語とかＣ＃言語からＲｕｓｔ言語に来ると、 メソッド・チェーン は見慣れないのよね～」

「　もう少し細かく見ていくと……☆」

「　サーバーというのは、　クライアントを　３つも　４つも、　多くを相手にしないといけない☆」

let server = listener.incoming().for_each( ☆ ).map_err( ☆ );
                     ^^^^^^^^^^^

「　ということは、この　.incoming()　という関数は、タイミングが重なれば
一度に　複数のクライアントを返すこともあるわけだぜ☆」

let server = listener.incoming().for_each( ☆ ).map_err( ☆ );
                                ^^^^^^^^^

「　だから後ろに、　for_each()　をチェーンしているな☆」

「　.for_each(☆)　の星の中に　クライアント１つに対応したクロージャを書けばいいわけかだぜ☆？」

「　理解が早いな……☆」

「　接続してきたのに失敗するクライアントとかいたら　どう対応すんの？」

let server = listener.incoming().for_each( ☆ ).map_err( ☆ );
                                               ^^^^^^^^

「　メソッド・チェーンのうしろの　.map_err()　でログでも出せだぜ☆」

「　なんで　.for_each()　のうしろに　.map_err()　が　くっついてんの？」

「　例えば　さっきの　りんご　の例だと　Result型を使って　↑上図のように　Ok　と　Err　に分けて
数を付けることができるわけだが……☆」

「　Ok　と　Err　に分かれていると考えろだぜ☆　すっきりしてるだろ☆
このような見方を　写像になっている　と呼んだりする☆　写像を英語で言うと　map☆」

let server = listener.incoming().for_each(|socket| {
        Ok(())
    })
    .map_err(|err| {
        println!("Error = {:?}", err);
    });

「　これで　↑このコード　の見方も見えてきただろ☆」

「　Ok(())　って何なの？　Ok　は震えてるの？」

「　Ok だが、返す中身はないということだぜ☆」

「　じゃあ　そこで　Err(5)　とか書いたら、制御はどこに返るの？
うしろの .map_err()　？」

「　どうなるんだろな☆？
fall through（フォールスルー）するんだろうか☆？　分かんないぜ☆」

「　しかし　お父ん☆？」

let server = listener.incoming().for_each( ☆ ).map_err( ☆ );
^^^^^^^^^^^^

「　なんで　返ってくるのは　server　なんだぜ☆？」

「　ぱっと見た感じ、それは　Future　（フューチャー; 先物）　なんだろうと想像が働くよな☆
フューチャーの説明は前の記事で、した☆」

tokio::run(server);

「　あとの方で　.run(server)　してれば、　Future　だと分かるぜ☆
同期処理と、非同期処理を、違いを気にせず書けるようにしてるから　run　という名前なんだろうが、
プロミス型の非同期処理では　await　と書く場合もある☆」

「　この調子で　むずかしいコードに目を慣らしていこうぜ☆」

handle client

Example: An Echo Server

let server = listener.incoming().for_each( ☆ ).map_err( ☆ );
                                          ^^^^

「　クライアント１つごとに　やる処理を調べていこうぜ☆」

let server = listener.incoming().for_each(|socket| {
  // split the socket stream into readable and writable parts
  let (reader, writer) = socket.split();
  // copy bytes from the reader into the writer
  let amount = io::copy(reader, writer);

  let msg = amount.then(|result| {
    match result {
      Ok((amount, _, _)) => println!("wrote {} bytes", amount),
      Err(e)             => println!("error: {}", e),
    }

    Ok(())
  });

  // spawn the task that handles the client connection socket on to the
  // tokio runtime. This means each client connection will be handled
  // concurrently
  tokio::spawn(msg);
  Ok(())
})

「　クライアントから送られてきた文字列を　そのまま返す、エコー・サーバー（Echo server）というサンプル・プログラムだぜ☆
解説していこう☆」

let server = listener.incoming().for_each(☆)

「　クロージャーを省くとこの形だぜ☆　この形はもう見たな☆」

|socket| {
  // split the socket stream into readable and writable parts
  let (reader, writer) = socket.split();
  // copy bytes from the reader into the writer
  let amount = io::copy(reader, writer);

  let msg = amount.then(|result| {
    match result {
      Ok((amount, _, _)) => println!("wrote {} bytes", amount),
      Err(e)             => println!("error: {}", e),
    }

    Ok(())
  });

  // spawn the task that handles the client connection socket on to the
  // tokio runtime. This means each client connection will be handled
  // concurrently
  tokio::spawn(msg);
  Ok(())
}

「　クロージャーはこうだぜ☆　この中身を解説していこう☆」

|socket| {☆}

「　クロージャーは　こういう形をしている☆
引数は　socket　だぜ☆」

  // split the socket stream into readable and writable parts
  let (reader, writer) = socket.split();
  // copy bytes from the reader into the writer
  let amount = io::copy(reader, writer);

  let msg = amount.then(|result| {
    match result {
      Ok((amount, _, _)) => println!("wrote {} bytes", amount),
      Err(e)             => println!("error: {}", e),
    }

    Ok(())
  });

  // spawn the task that handles the client connection socket on to the
  // tokio runtime. This means each client connection will be handled
  // concurrently
  tokio::spawn(msg);
  Ok(())

「　クロージャーの本体は　↑こうだな☆」

  let (reader, writer) = socket.split();

「　ソケットは、読取ストリームと　書込ストリームの２つに分けるのが定番なようだぜ☆　真似しよう☆」

  let amount = io::copy(reader, writer);

「　io::copy()　というのは　読取ストリームから入ってきたものを、書込ストリームにそのまま　送り出すようだぜ☆
これが　エコー　の仕組みだな☆
で、返り値が　amount　だが、これは　非同期処理の　Future　と見るべきだろう☆」

「　Function　の返り値なのか、　非同期処理の　Future　なのか、
どっちが返ってきているのか　どうやって見分けるんだぜ☆？」

「　よい見分け方はない☆　見分けにくいよな☆
これが改善されるのか、そのまま　言語実装になるのかは　今の時点では分からないぜ☆」

「　プログラム言語は　型推論　が自動で行われるのが流行りですから、
Functionの返り値なのか、Futureなのかは　見分けられなくなるように進化　していく流れなのよ。
人類はボトルネックなの。　コンピューターが見分けてくれるから、人類は下手に見分けようとしなくていいのよ」

「　Functionの返り値と、Future　の使い勝手が　合同　なら　見分けられなくなるように進化　するのはいいが、
使い勝手は異なるから　結局　見分けないと使えないんだろ☆？　ややこしいぜ☆」

「　Ｃ言語から　Ｒｕｓｔ言語に来た人なら、何段階か頭をギアチェンジしないといけないよな☆
そのアップデートができないと　ここで終わりだぜ☆」

「　プログラムの文法は　こういうものだ、と　頭の中で形ができあがっちゃうと　定型文を組み合わせるだけの文学になってしまうのよ。
文法を発明する　数学　の頭をしていれば　コロっと　より新しいものを取り込める普遍的な記法　の方へ乗り換えられるのよ」

「　そんな説明では　分からないんで☆」

let msg = amount.then(|result| {
    match result {
      Ok((amount, _, _)) => println!("wrote {} bytes", amount),
      Err(e)             => println!("error: {}", e),
    }

    Ok(())
  });

「　続きのコードを見てみよう☆」

let msg = amount.then(☆);

「　クロージャーを省くと　↑こんな形をしている☆
amount　は　Future　だとしよう☆　.then()　というメソッドは、何かあったときに
その引数に渡されたクロージャーを実行するのだろう☆
Tokioクレートの話題をしていて、クロージャーを使ってるなら　非同期処理っぽいよな☆」

|result| {
    match result {
      Ok((amount, _, _)) => println!("wrote {} bytes", amount),
      Err(e)             => println!("error: {}", e),
    }

    Ok(())
  }

「　クロージャーだけ見ると　↑こんな形をしている☆
Result型の result 引数が渡ってくるんで、目新しいのは match構文で　スイッチしてるところだけだな☆」

    match result {
      Ok((amount, _, _)) => println!("wrote {} bytes", amount),
      Err(e)             => println!("error: {}", e),
    }

「　Rust言語では switch文が無いので　その代わりに match 構文を使う☆
Ok の方は値が　タプルになっていて　(amount, _, _)　と書いてある☆　まあ、Result型の値は様々だからな☆
見たところ　何かの量を表す数　が入ってそうだぜ☆」

「　なにをやりたがっているのか　想像しながら　プログラムを読まなくちゃいけないのね。
大昔のアセンブリ言語に先祖返りしたみたい」

「　で、エラーもメッセージを表示しているだけなんで……☆」

    Ok(())

「　このクロージャーは　必ず　値なしのOk　で終わるな☆」

let msg = amount.then(☆);
^^^^^^^^^

「　Ok は .then( ) メソッドの中のどこかへ返してるんだろ☆
そして　.then( ) メソッドは msg を返している☆　これも　Future　かだぜ☆？」

  tokio::spawn(msg);
  Ok(())

「　msg　は　::spawn( )　メソッドに渡してるな☆　スレッドを作るメソッドなわけだが、
msg　は　コールバック関数なんだろうか☆？」

「　tokio::run( )　じゃなくて　tokio::spawn( )　を使うの？」

「　run( )　は　開始の合図みたいなもんで、
それ以外は　開始したあとに　どう動くかといった　パイプラインの設計図を作っているのだろう☆」

「　同期のプログラムを組んでいるのか、非同期処理のパイプラインの設計図を作っているのか、見た目から分からん……☆」

「　同期処理と、非同期処理を　同じ見た目で書ける　というのも　ウリ　なんだろうけどな☆」

Future を自作しようぜ☆（＾～＾）

Implementing futures

「　↑フーチャーを使うだけでなく、自分で書く方法も説明されている☆」

trait Future {
    type Item;
    type Error;
    fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error>;
}

「　↑覚えることは３つしかないぜ☆」

「　Item　とかいう　どうとでも受け取れる名前を付けられると　何なのか　わかんないよな☆」

「　しかも　type　だし」

「　その　Item　も、 Error　も、 poll メソッドで使う型なわけなんで、
よく見てしまえば、結局　poll　（ポール）メソッド１つ　定義しているだけだぜ☆」

    fn poll(&mut self) -> ☆;

「　poll メソッドの引数は　無いようなもんなんで……☆」

    ☆ -> Result<Async<Self::Item>, Self::Error>;

「　じゃあ　poll メソッドで一番むずかしいのは　返り値　ということになるが……☆」

    Result<☆, ☆>

「　Result　型は前に説明したな☆」

    Async<Self::Item>,
    Self::Error

「　すると Ok のとき　Async<Self::Item>　型を一緒に返すし、
Err のとき　Self::Error　型を一緒に返すということだな☆」

「　Self::Item　型とか、　Self::Error　型とか聞いたことないぞ、何だ、というと……☆」

trait Future {
    type Item;
    type Error;
    fn ☆;
}

「　Future　トレイトで　Item　と、　Error　の型は指定できるな☆」

「　じゃあ　Future　って、その成分の１００％は　poll メソッドの返り値　ってことなの？
じゃあ　Future　は、返り値なの？」

「　まあ、まさに　そういうことなのだろう☆」

「　Async< >　なんていう型　聞いたことないんだが☆」

Enum futures::Async

「　↑聞いたことなければ、自分で調べろだぜ☆」

pub enum Async<T> {
    Ready(T),
    NotReady,
}

「　↑ Ready 準備ができたか、 NotReady できてないかの２択だな☆」

「　Ready の対義語が NotReady ってのも　なんか　ダサいわねぇ」

「　じゃあ　poll メソッドは、準備できたItem、準備できてない、エラーError　の３パターンの終わり方があるぜ☆」

「　そういうことだな☆」

「　Future って、それだけなの？」

「　それだけだな☆」

「　実装サンプルを見てみようぜ☆？」

Implementing futures

extern crate futures;

// `Poll` is a type alias for `Result<Async<T>, E>`
use futures::{Future, Async, Poll};

struct HelloWorld;

impl Future for HelloWorld {
    type Item = String;
    type Error = ();

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        Ok(Async::Ready("hello world".to_string()))
    }
}

「　糖衣構文みたいな技を使ってるわよ。 Result<Async<☆>,☆>　なんて書かずに　Poll<☆,☆>　と書けばいいみたいよ」

「　それぐらい簡単にしてくれなければ　書きたくないよな☆」

「　なんか　簡単な話しなんだな☆」

「　このコードは　常に　Ready 準備できていて、 hello world を返すぜ☆
じゃあ　どうやって　この非同期処理を実行するかだな☆」

extern crate futures;

use futures::{Future, Async, Poll};
use std::fmt;

struct Display<T>(T);

impl<T> Future for Display<T>
where
    T: Future,
    T::Item: fmt::Display,
{
    type Item = ();
    type Error = T::Error;

    fn poll(&mut self) -> Poll<(), T::Error> {
        let value = match self.0.poll() {
            Ok(Async::Ready(value)) => value,
            Ok(Async::NotReady) => return Ok(Async::NotReady),
            Err(err) => return Err(err),
        };

        println!("{}", value);
        Ok(Async::Ready(()))
    }
}

「　Display 構造体が poll メソッドを持っているような例が　サンプルで示されているぜ☆」

「　詳しく☆」

extern crate futures;

use futures::{Future, Async, Poll};
use std::fmt;

「　std::fmt::Display　トレイトを利用するわけだな☆」

Trait std::fmt::Display

pub trait Display {
    fn fmt(&self, f: &mut Formatter) -> Result<(), Error>;
}

「　この fmt メソッドは例えば　println!("{}", value);　みたいなコードがあるとき、
"{}"　みたいな文字列をフォーマットと呼ぶが、この波括弧を　別の文字列に置換してくれる働きをするぜ☆」

「　この Display トレイトを、非同期処理で利用できるようにしようというサンプルだぜ☆」

struct Display<T>(T);

impl<T> Future for Display<T>
where
    T: Future,
    T::Item: fmt::Display,
{
    ☆
}

「　分かりづらいが、あとで Display( ☆ ) みたいなコンストラクターみたいな書き方ができる構造体を作っている☆
その　Display構造体に対して、　T が Future で、 FutureのItemが std::fmt::Display のとき、トレイトを実装する……、
みたいなことが書いてある☆　さらに本体を見てみよう☆」

    type Item = ();
    type Error = T::Error;

    fn poll(&mut self) -> Poll<(), T::Error> {
        ☆
    }

「　これは　Feature　トレイトの実装だな☆　準備ができているときは　() 値はなしで、
エラーのときは多分 FutureのErrorか☆」

        let value = match self.0.poll() {
            Ok(Async::Ready(value)) => value,
            Ok(Async::NotReady) => return Ok(Async::NotReady),
            Err(err) => return Err(err),
        };

        println!("{}", value);
        Ok(Async::Ready(()))

「　self.0　という書き方が嫌らしいが　トレイトを実装したオブジェクトの最初のプロパティ　という意味かだぜ☆？
その最初のプロパティも　.poll() メソッドを持っている……☆？
match 構文で３行並んでいるのは　準備できている、準備できていない、エラー　の３つだな☆」

「　HelloWorld トレイトは必ず　準備できているんで、その他の分岐は適当に作ってるんだろう☆」

extern crate tokio;

let future = Display(HelloWorld);
tokio::run(future);

「　あとは　tokioクレートで　非同期処理を　run　するだけだな☆」

「　必ず準備できているような非同期処理なら、糖衣構文があるみたいよ？」

#[macro_use]
extern crate futures;

use futures::{Future, Async, Poll};
use std::fmt;

struct Display<T>(T);

impl<T> Future for Display<T>
where
    T: Future,
    T::Item: fmt::Display,
{
    type Item = ();
    type Error = T::Error;

    fn poll(&mut self) -> Poll<(), T::Error> {
        let value = try_ready!(self.0.poll());
        println!("{}", value);
        Ok(Async::Ready(()))
    }
}

「　try_ready!( ) というマクロがあるのかだぜ☆」

「　結構長くなったんで、続きは次の記事でだぜ☆」

「　まだだぜ☆　完全なるソースを示せだぜ☆」

Implementing futures - example source

Cargo.toml

[dependencies]
tokio = "0.1"
futures = "0.1.26"

「　クレートは自分で crates.io から検索しような☆」

main.rs

//!
//! cargo new ep1
//! cd C:\Users\むずでょ\OneDrive\ドキュメント\practice-rust\tokio\impl-futu
//! cargo check
//! cargo build
//! cargo run
//! 
//! [Implementing futures](https://tokio.rs/docs/futures/basic/)
//!

extern crate tokio;
extern crate futures;

use futures::{Future, Async, Poll, try_ready};
use std::fmt;


struct HelloWorld;

impl Future for HelloWorld {
    type Item = String;
    type Error = ();

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        Ok(Async::Ready("hello world".to_string()))
    }
}


struct Display<T>(T);

impl<T> Future for Display<T>
where
    T: Future,
    T::Item: fmt::Display,
{
    type Item = ();
    type Error = T::Error;

    fn poll(&mut self) -> Poll<(), T::Error> {
        let value = try_ready!(self.0.poll());
        println!("{}", value);
        Ok(Async::Ready(()))
    }
}

fn main() {
    let future = Display(HelloWorld);
    tokio::run(future);
}

「　ソースはこんな感じだぜ☆」

＜次回に続く＞

ツイッターでシェア

みんなに共有、忘れないようにメモ