非同期処理

ひとまず Tokio のチュートリアルをやっていく

https://zenn.dev/magurotuna/books/tokio-tutorial-ja

https://tokio.rs/tokio/tutorial

リポジトリ作った

https://github.com/nukopy/tokio-tutorial

References

RustのTokioで非同期とグリーンスレッドを理解する

https://zenn.dev/tfutada/articles/5e87d6e7131e8e

Rustのスレッドとチャネルと共有メモリの話

https://zenn.dev/tfutada/articles/16766e3b4560db

Mini-Redis Tutorialからはじめるtokio

https://blog.ymgyt.io/entry/mini_redis_tutorial_to_get_started_with_tokio/

並行性と並列性

並行（concurrency）と並列（parallelism）は同じものではありません。もしあなたが、2つのタスクを交互に行うのなら、それはそれらのタスクを「並行して」こなしているということになりますが、「並列」ではありません。これを「並列」にしたいのであれば、2人の人間が必要で、それぞれのタスクにそれぞれの人間を専任させることになるでしょう。

Tokio を使うことによる利点の1つは、非同期コードを書くことによって、普通のスレッドを利用して多くのタスクを並列に処理するのではなく、「並行に」処理することができるようになる、という点です。実際、Tokio を使えば、たとえ単一のスレッドであったとしても、多くのタスクを並行して実行することができるのです！

Tokio の非同期タスクのスレッド間の移動

これどういう仕組み？<s>main 実行時のスレッドってどうなってるの？</s> ←何いってんだおまえ。スレッドについて理解できていなかったときに出た疑問。解消済み。

spawn されたタスクは、spawn されたスレッドと同一のスレッド上で実行されることもあれば、ランタイム上の異なるスレッドで実行されることもあります。spawn されたあとにスレッド間を行き来することもあります。

https://zenn.dev/magurotuna/books/tokio-tutorial-ja/viewer/spawning#タスク

非同期タスクの生成に必要なリソース

なるほど。Go の goroutine みたなもの？
（Go の goroutine はグリーンスレッドって言って良い？）

Tokio におけるタスクはとても軽量です。内部的には、1回のアロケーションしか必要としませんし、64バイトのメモリしか消費しません。アプリケーションは気軽に、数百万とまではいかないものの、数千のタスクを spawn することができます。

参照カウント Rc, Arc の基本

参照カウントとは

参照カウントはメモリ管理のテクニックの一つで、特定のリソースへの「参照」または「ポインタ」の数を追跡する。リソースが最初に作成されたとき、カウントは 1 から始まる。新しい参照が作成されるたびにカウントは増加し、参照がドロップ（スコープを抜けるなど）されるたびにカウントは減少する。カウントが 0 になると、そのリソースはメモリから解放される。

Rustの Rc と Arc

Rust には、参照カウントのための2つの主要な型、Rc と Arc がある。

1. Rc (Reference Counted)
   • Rcは、シングルスレッド環境での参照カウントを提供する
   • 複数の所有者を持つことができるが、これはシングルスレッド環境での使用でのみ安全
2. Arc (Atomic Reference Counted)
   • Arcは、複数のスレッド間でデータを共有するためのアトミックな参照カウントを提供する
   • 複数のスレッドから同時にアクセスされる可能性があるデータに対して安全

基本的な使い方

1. Rc の使用

   use std::rc::Rc;
   
   let foo = Rc::new(vec![1.0, 2.0, 3.0]);
   
   // この時点で、参照カウントは 1
   
   let a = foo.clone();
   let b = foo.clone();
   
   // この時点で、参照カウントは 3
   
   // a, b, foo はすべて同じデータにアクセスする
   println!("{:?}", a);
   

2. Arc の使用:

   use std::sync::Arc;
   use std::thread;
   
   let foo = Arc::new(vec![1.0, 2.0, 3.0]);
   
   // この時点で、参照カウントは 1    
   
   let a = foo.clone();
   let b = foo.clone();
   
   // この時点で、参照カウントは 3
   
   thread::spawn(move || {
       println!("{:?}", a);
   }).join().unwrap();
   
   thread::spawn(move || {
       println!("{:?}", b);
   }).join().unwrap();
   

Rc と Arc は、メソッドや関数がデータの所有権を持つ必要がない場合や、複数の場所でデータを共有する必要がある場合に特に役立つ。ただし、これらの型を使用する際は、参照カウントのオーバーヘッドを考慮する必要がある。また、Rc や Arc でラップされたデータは、データが変更される可能性がある場合、RefCell や Mutex などの内部可変性を提供するメカニズムと組み合わせて使用する必要がある。

Into トレイト

.into() は Rust において非常に便利なメソッドで、Into トレイトに関連している。このトレイトとその関連トレイト、そしてその使用方法について説明する。

1. Into トレイト

Into トレイトは、ある型を別の型に変換するための機能を提供する。これは非常に一般的なトレイトで、多くの標準ライブラリの型に実装されている。

例:

let s: String = "hello".into(); // &str を String に変換

2. From トレイト

From トレイトは Into の逆。From を実装することで、自動的に Into も実装される。

例:

let s = String::from("hello"); // &str から String への変換

3. into の使用

下記のコードの中で、value.into() は、value の型（この場合は &str）を、client.set メソッドが期待する型に変換している。具体的には、&str を String などの所有権を持つ文字列型に変換している。

use mini_redis::{client, Result};

#[tokio::main]
async fn main() -> Result<()> {
    // Open a connection to the mini-redis address.
    let addr = "127.0.0.1:6379";

    // Set the key "hello" with value "world"
    let key = "hello";
    let value = "world";
    println!("Setting key {} to {}.", key, value);
    let mut client = client::connect(addr).await?;
    client.set(key, value.into()).await?;

    // Get key "hello"
    let mut client = client::connect(addr).await?;
    let result = client.get(key).await?;

    println!("Got value from mini-redis server; result = {:?}", result);

    Ok(())
}

4. なぜ into が便利か

• 型推論
  • Rust の型推論システムと組み合わせることで、into を使うときに変換先の型を明示的に指定する必要がなくなる。コンパイラがコンテキストから適切な型を推論してくれる。
• コードの簡潔性
  • 変換関数を都度書く代わりに、一貫したメソッドを使用して様々な型の間の変換を行うことができる。

Into や From トレイトは、カスタム型に対しても実装することができる。これにより、異なるカスタム型間での変換を簡潔に行うことができる。

シャーディング

Rust でのシンプルなシャーディングの例

https://zenn.dev/magurotuna/books/tokio-tutorial-ja/viewer/shared_state#タスク、スレッド、競合

シャーディングにおける「ホットスポット」

ChatGPT: なるほど、じゃあキーのハッシュ値によっては特定のシャードへの insert が偏ることもありえるんだね。

はい、その通りです。キーのハッシュ値の分布が一様でない場合、特定のシャードに対するアクセスや挿入が偏る可能性があります。これは「ホットスポット」として知られ、一部のシャードが過度に使用される一方で、他のシャードがほとんど使用されないという状況を引き起こすことがあります。

ホットスポットが発生すると、次のような問題が起こる可能性があります：

1. パフォーマンスの低下：特定のシャードが過度に使用されると、そのシャードに対するアクセスが頻繁にブロックされる可能性があります。
2. リソースの非効率的な使用：一部のシャードが過度に使用される一方で、他のシャードが十分に活用されない場合、リソースの非効率的な使用となります。

ホットスポットを避けるための方法：

1. 良いハッシュ関数の使用：ハッシュ関数が一様な分布のハッシュ値を生成することで、シャードへのデータの分散を最適化することができます。
2. 動的リバランシング：シャード間のデータの分布を定期的に監視し、必要に応じてデータを再分散することで、偏りを軽減することができます。
3. シャーディング戦略の再評価：キーのハッシュ値だけでなく、他の属性や戦略を考慮してシャーディングを行うことで、データの分散を最適化することができます。

シャーディングは非常に強力なテクニックですが、設計や実装の段階で考慮すべき多くの要因があります。適切なシャーディング戦略を選択し、適切にモニタリングと調整を行うことで、データの効率的な管理と高いパフォーマンスを維持することができます。

Tokio のチャンネルの種類

https://docs.rs/tokio/1.32.0/tokio/sync/index.html

https://zenn.dev/magurotuna/books/tokio-tutorial-ja/viewer/channels#tokio-のチャネルプリミティブ

• mpsc
  • multi-producer, single-consumer 型のチャネル。たくさんの値を送ることができる。
• oneshot
  • single-producer, single-consumer 型のチャネル。1つの値を送ることができる。
• broadcast
  • multi-producer, multi-consumer 型のチャネル。受信側はすべての値を見ることができる。
• watch
  • single-producer, multi-consumer 型のチャネル。たくさんの値を送ることができるが、履歴は残らない。受信側は最新の値のみを見ることができる。

非同期 I/O

https://zenn.dev/magurotuna/books/tokio-tutorial-ja/viewer/io

非同期タスクにおけるメモリ使用量の注意

これ重要。

https://zenn.dev/magurotuna/books/tokio-tutorial-ja/viewer/io#バッファをアロケートする

分かりにくい文章になってしまったと思うので補足させてください。
ここでは、バッファの確保を [u8; 1024] （スタック上に確保される配列型）で行うよりも Vec<u8> （ヒープ上に確保されるベクタ）で行ったほうが効率が良い、ということを説明しています。その理由として、以下の2つが挙げられています。

a) スタック配列だと .await をまたぐときに全体のムーブが必要になること（ベクタであればヒープへのポインタ + 補助的な僅かなデータをムーブするだけで済む）
b) スタック配列の場合、タスクを表す構造体がバッファデータをすべて含むことになるため、構造体サイズが大きくなってしまうこと（ベクタであればヒープへのポインタ + 補助的な僅かなデータ を含むだけでよく、軽量に保てる）

非同期プログラミングとメモリの確保場所（スタック vs ヒープ）に関する高度なトピック。特にRust の非同期コードの文脈において、どのようにデータがメモリ上で配置され、どのようにそれが非同期タスクの性能やメモリ使用量に影響するかについて説明している。

簡単に言えば、非同期タスク内で大きなデータ構造を持つと、その非同期タスクが await したりスレッド間を跨ぐとき、データ構造全体をコピーする必要があるため、パフォーマンスの低下やメモリの無駄遣いを引き起こす可能性がある。このコピーオーバーヘッドは、データがスタック上に確保されている場合、特に顕著になる。

逆に、ヒープ上にデータを確保すると、非同期の「境界」をまたぐときにデータの実際の内容をコピーする必要はない。代わりに、ヒープ上のデータへのポインタだけがコピーされるため、オーバーヘッドが少なくなる。つまり、スタック領域でなくバッファ専用のアロケーションを都度行ったほうが効率が良い。

サンプルコード

引用元

• ポイント
  • バッファの確保を [u8; 1024] （スタック上に確保される配列型）で行うよりも Vec<u8> （ヒープ上に確保されるベクタ）で行ったほうが効率が良い。つまり、このケースではスタック領域でなくバッファ専用のアロケーションを都度行ったほうが効率が良い。

use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> io::Result<()> {
    let mut listener = TcpListener::bind("127.0.0.1:6142").await.unwrap();

    loop {
        let (mut socket, _) = listener.accept().await?;

        tokio::spawn(async move {
            let mut buf = vec![0; 1024];

            loop {
                match socket.read(&mut buf).await {
                    // `Ok(0)` が返ってきたらリモート側が閉じられたことを意味する
                    Ok(0) => return,
                    Ok(n) => {
                        // データをソケットへとコピーする
                        if socket.write_all(&buf[..n]).await.is_err() {
                            // 予期しないソケットエラーが発生した場合。
                            // ここで何かできることはさほどないので、処理を停止する
                            return;
                        }
                    }
                    Err(_) => {
                        // 予期しないソケットエラーが発生した場合。
                        // ここで何かできることはさほどないので、処理を停止する
                        return;
                    }
                }
            }
        });
    }
}

Redis で使用される通信プロトコル RESP（Redis Serialization Protocol）

https://redis.io/docs/reference/protocol-spec/

概要

（DeepL で翻訳）

Redis シリアライゼーションプロトコル（RESP）は、クライアントが実装するワイヤプロトコルである。

Redis サーバーと通信するために、Redis クライアントは REdis Serialization Protocol（RESP）と呼ばれるプロトコルを使用する。このプロトコルは Redis 専用に設計されたものだが、他のクライアント・サーバー・ソフトウェアにも使用できる。

RESPは、以下の考慮事項の妥協点である：

• 実装が簡単
• 解析が速い
• 人間が読める

RESP は、整数、文字列、配列などのさまざまなデータ型をシリアライズ（メモリ→ネットワーク）することができる。エラー固有の型もある。クライアントはリクエストを文字列の配列として Redis サーバに送信する。配列の中身は、サーバーが実行すべきコマンドとその引数である。サーバーの応答型はコマンド固有。

RESP はバイナリセーフで、バルクデータの転送に前置長を使用するため、あるプロセスから別のプロセスへ転送されるバルクデータを処理する必要はない。

RESP は Redis クライアントに実装すべきプロトコルである。

注意：
RESP プロトコルは Redis サーバ・クライアント間の通信に使用されるものである。Redis Cluster はノード間でメッセージを交換するために RESP とは異なるバイナリプロトコルを使用する。

RESP の整理

• RESP, Redis Serialization Protocol
  • Redis サーバ・クライアント間の通信に使用される Redis 専用のプロトコル。
  • TCP/IP におけるアプリケーション層のプロトコルであり、主に TCP 上で使用される。
  • リクエスト / レスポンスモデル

RESP で使用されるデータ型（mini-redis で Frame として定義されているもの）

RESP の主なデータ型（フレーム）：

1. 単純文字列 (Simple Strings)
   • + で始まる
   • エラーではない文字列の情報を示す
   • 例: +OK\r\n は、多くの Redis コマンドの成功を示す標準的な応答
2. エラー (Errors)
   • - で始まる
   • エラーメッセージを示す
   • 例: -ERR unknown command 'foobar'\r\n
3. 整数 (Integers)
   • : で始まる
   • 32ビット符号付き整数を示す
   • 例: :1000\r\n
4. バルク文字列 (Bulk Strings)
   • $ で始まり、バイト数が続く
   • 特定のバイト数の文字列を示す
   • 例: $6\r\nfoobar\r\n は "foobar" という6文字の文字列を示します
   • バルク文字列が存在しない場合は $-1\r\n となります
5. 配列 (Arrays)
   • * で始まり、続く要素数が続く
   • その後、要素ごとにRESPデータ型が続く
   • 例: *2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n は "foo" と "bar" という2つのバルク文字列からなる配列を示します
   • 配列が存在しない場合は *-1\r\n となります

RESPは、クライアントとサーバ間でコマンドを交換するためのプロトコルとして設計されている。クライアントは、コマンドとその引数を配列としてサーバに送信する。サーバは、応答として単純文字列、エラー、整数、バルク文字列、または配列を返すことができる。

このような形式は、Redis のようなキー-バリュー型のデータストアに特に適しており、シンプルさと効率性を兼ね備えている。

Rust の非同期ランタイムモデル

https://tokio.rs/tokio/tutorial/async

https://zenn.dev/magurotuna/books/tokio-tutorial-ja/viewer/async_in_depth