🤹🏽 🛳️ 🕰️ 錆の詳細：スケーラブルなチャットをゼロから作成する、パート1 🙍🏽 ⛹🏽 🧒🏿

パート1：WebSocketを実装します。はじめに

この一連の記事では、リアルタイムで機能するスケーラブルなチャットを作成するプロセスについて説明します。

このレビューの目的は、Rustプログラミング言語が実際に急速に普及している基礎の基礎を段階的に研究することであり、システムインターフェイスを網羅しています。

最初の部分では、環境の初期設定と最も単純なWebSocketサーバーの実装について検討します。記事の技術的な詳細を理解するには、Rust言語の経験は必要ありませんが、システムAPI（POSIX）とC / C ++の基本的な知識は不必要ではありません。読み始める前に、少し時間をとって（そしてコーヒーも）取りましょう-この記事では、すべてを可能な限り詳細に説明しているため、非常に長くなります。

1 Rust-選択の理由

Rustプログラミング言語に興味を持つようになったのは、長年にわたるシステムプログラミングへの情熱であり、面白くもありながら非常に複雑です。

そして、おそらく、ここで最も難しい問題は、メモリを使用した安全な作業と呼ぶことができます。バッファオーバーフロー、メモリリーク、二重メモリ割り当て解除、参照のダングリング、すでに解放されたメモリへのポインタの逆参照など、多くのバグを引き起こすのはメモリの不適切な動作です。また、このようなエラーは深刻なセキュリティ問題を伴うことがあります。たとえば、OpenSSLの悪名高いバグHeartbleedの原因は、メモリのずさんな処理にすぎません。そして、これは氷山の一角にすぎません。私たちが毎日使用しているソフトウェアにそのようなギャップがいくつ隠されているのかは誰にもわかりません。

C ++では、このような問題を解決するためのいくつかの方法が考案されました。たとえば、スマートポインターの使用^[1]やスタック上の割り当て^[2]などです。残念ながら、そのようなアプローチを使用しても、「足を撃つ」と呼ばれるものがまだ存在する可能性があります-バッファーの境界を越えるか、常に使用可能なメモリを操作するための低レベル関数を使用します。

つまり、言語レベルでは、このようなプラクティスを適用するための前提条件はありません。代わりに、「優れた開発者」は常に自分自身を使用し、間違いを犯さないと考えられています。ただし、大量のコードを手動で完全にチェックすることはできないため、コードにこのような重大な問題が存在することは、開発者のレベルとはまったく関係ないと考えています。これはコンピューターのタスクです。ある程度までは、静的解析ツールがここで役立ちますが、すべてではなく、常に使用するわけではありません。

このため、メモリの操作で問題を取り除く別の基本的な方法があります。ガベージコレクションは、コンピュータサイエンスの別の複雑な知識の領域です。ほとんどすべての最新の言語と仮想マシンには何らかの形の自動ガベージコレクションがあり、ほとんどの場合これは非常に優れたソリューションであるという事実にもかかわらず、欠点があります。まず、自動ガベージコレクタは理解と実装が困難です^] 。第二に、ガベージコレクションの使用は、未使用のメモリを解放するための一時停止を意味します^[4] 。これは通常、高負荷のアプリケーションで遅延を減らすための微調整の必要性を伴います。

Rustにはこの問題に対する別のアプローチがあります。黄金の平均とは、追加のメモリやプロセッサ時間を必要とせず、各ステップの自己追跡が不要なメモリとリソースの自動解放です。これは所有権と借用の概念の適用によって達成されます。

言語は、各値が所有者を 1人だけ持つことができるというステートメントに基づいています。つまり、メモリの特定の領域を指す可変変数は1つしか存在できません。

let foo = vec![1, 2, 3]; //     (),   1, 2,  3, //       `foo`. let bar = foo; //     `bar`. //          `foo`, //      "" - ..,    .

このアプローチには興味深い結果があります：値は1つの変数のみに関連付けられているため、変数がスコープを出ると、この値に関連付けられたリソース（メモリ、ファイル記述子、ソケットなど）が自動的に解放されます（中括弧内のコードブロックによって設定されます）、 {

および}

）。

そのような人為的な制限は不必要で過度に複雑に思えるかもしれませんが、慎重に考えれば、これは概して、Rustの「キラー機能」であり、実用的な理由のためだけに現れました。このアプローチにより、C / C ++で記述された低レベルコードの有効性を維持しながら、Rustを高レベル言語のように見せることができます。

しかし、すべての興味深い機能にもかかわらず、最近までRustには深刻な欠陥がありました-たとえば、互換性を保証できない非常に不安定なAPIなどです。しかし、この言語の作成者はほぼ10年で大きな進歩を遂げており^[5] 、現在、安定バージョン1.0のリリースにより、言語は実際のプロジェクトで実践できるように進化しています。

2ゴール

私は、現実の世界で比較的簡単なプロジェクトを開発しながら、新しい言語と概念を学ぶことを好みます。したがって、言語の可能性は、必要になったときに正確に研究されます。 Rustを探索するプロジェクトとして、Chat Rouletteなどの匿名チャットサービスを選択しました。私の意見では、これはチャットがサーバーからの短い応答時間を通常要求し、多数の同時接続を意味するという理由から適切な選択です。私たちは数千を頼りにします-したがって、実際の環境でRustで書かれたプログラムのメモリ消費とパフォーマンスを見ることができます。

最終結果は、さまざまなクラウドホスティングサービスにサーバーを展開するためのスクリプトを含むバイナリプログラムファイルになります。

しかし、コードを書き始める前に、I / Oでいくつかのポイントを説明するために少し余談をする必要があります。それを適切に操作することがネットワークサービスの開発における重要なポイントだからです。

3 I / Oオプション

タスクを完了するには、サービスがネットワークソケットを介してデータを送受信する必要があります。

一見、タスクは簡単ですが、実際には、さまざまな複雑さやさまざまな有効性を解決するための多くの方法があります。それらの主な違いはロックへのアプローチにあります。ここでの標準的なプラクティスは、新しいデータがソケットに到着するのを待っている間にプロセッサを停止することです。

他のユーザーをブロックする1人のユーザー向けのサービスを構築することはできないため、それらを互いに何らかの形で分離する必要があります。典型的な解決策は、ユーザーごとに個別のスレッドを作成することです。したがって、プロセス全体がブロックされるのではなく、そのスレッドの1つだけがブロックされます。このアプローチの短所は、比較的単純であるにもかかわらず、メモリ消費が増加することです。各スレッドは、作成されるとスタック用にメモリの一部を予約します^[6] 。さらに、実行コンテキストを切り替える必要があるため、問題は複雑です。最新のサーバープロセッサには通常8から16コアがあり、ハードウェアが許可するよりも多くのスレッドを作成すると、OSスケジューラーは十分な速度でタスクスイッチングに対処しなくなります。

したがって、マルチスレッドプログラムを多数の接続にスケーリングすることは非常に困難な場合があり、この場合、数千の同時接続ユーザーを計画しているため、まったく合理的ではありません。最終的には、Habraeffectに備える必要があります！

4イベントループ

入出力の効率的な作業のために、イベント処理サイクルに基づいた多重化システムAPIを使用します。 Linuxカーネル^[7]にはこのためのepollメカニズムがあり、FreeBSDおよびOS Xにはkqueue ^[8]があります。

これらのAPIはどちらもかなり似た方法で配置されており、一般的な考え方は単純です。新しいデータがネットワーク経由でソケットに届くのを待つのではなく、 ソケットに到着したバイトを通知するように依頼します。

イベントの形式のアラートは一般的なサイクルに入りますが、この場合はブロッカーとして機能します。つまり、新しいデータの数千のソケットを絶えずチェックするのではなく、ソケットがそれについて通知するのを待つだけです-そして、接続されたユーザーは非常に頻繁にスタンバイモードになり、何も送信せず、受信しています。これは、WebSocketを使用するアプリケーションに特に当てはまります。さらに、非同期I / Oを使用すると、オーバーヘッドはほとんどありません。メモリに保存する必要があるのは、ソケットファイル記述子とクライアントの状態だけです（チャットの場合、これは接続ごとに数百バイトです）。

このアプローチの興味深い機能は、ネットワーク接続だけでなく、たとえばディスクからファイルを読み取るために非同期I / Oを使用できることです。イベントループは、あらゆるタイプのファイル記述子を受け入れます（* NIXの世界のソケットはまさにそれです）。

Node.jsのイベントループとRubyのEventMachine gemはまったく同じように機能します。

同じことが、非同期I / O ^[9]のみを使用するnginx Webサーバーの場合にも当てはまります。

5はじめに

追加のテキストは、Rustが既にインストールされていることを意味します。まだない場合は、公式Webサイトのドキュメントに従ってください。

Rustの標準配信には、Maven、Composer、npm、またはrakeに類似した機能を実行するcargo

と呼ばれるプログラムがあります-アプリケーションの依存関係を管理し、プロジェクトをビルドし、テストを実行し、最も重要なことは、新しいプロジェクトの作成プロセスを簡素化します。

これはまさに今必要なものです。そこでターミナルを開いてこのコマンドを入力してみましょう。

 cargo new chat --bin

--bin

引数は、ライブラリではなくスタートアップアプリケーションを作成するようCargoに指示します。

その結果、2つのファイルが作成されます。

 Cargo.toml src/main.rs

Cargo.toml

は、説明とプロジェクトの依存関係へのリンクが含まれています（JavaScriptのpackage.json

と同様）。

src/main.rs

はメインソースファイルであり、プログラムへのエントリポイントです。

最初に他に何も必要ないため、1つのコマンドでプログラムをコンパイルして実行することができますcargo run

。同じコマンドは、もしあれば、コードにエラーを表示します。

あなたが幸せなEmacsユーザーなら、彼はCargoと「すぐに使える」互換性があることを喜んで知るでしょう-MELPAリポジトリからrust-mode

パッケージをインストールし、 cargo build

を起動するようにコンパイルコマンドを設定cargo build

です。

6 Rustでのイベント処理

理論から実践に移ります。新しいメッセージが表示されるのを待つ最も単純なイベントループを実行してみましょう。これを行うために、さまざまなシステムAPIを手動で接続する必要はありません- 「 Metal IO 」またはmioと呼ばれる非同期I / Oを操作するために既存のライブラリを使用するだけです。

覚えているように、Cargoプログラムは依存関係を扱います。 crates.ioリポジトリからライブラリをダウンロードしますが、さらにGitリポジトリから直接それらを取得することができます-この機能は、まだパッケージリポジトリにロードされていないライブラリの最新バージョンを使用する必要がある場合に役立ちます。

この記事の執筆時点では、 mio

のリポジトリには古いバージョン0.3しかありません。開発中のバージョン0.4では、多くの便利な変更が行われており、古いバージョンとも互換性がありません。したがって、GitHubを介して直接接続し、そのような行をCargo.toml

追加しCargo.toml

。

 [dependencies.mio] git = "https://github.com/carllerche/mio"

プロジェクトの説明で依存関係を定義した後、インポートをmain.rs

追加しmain.rs

。

 extern crate mio; use mio::*;

mio

使用mio

非常に簡単です。まず、 EventLoop::new()

関数を呼び出してイベントループを作成しましょう。ただし、空のループは役に立たないので、すぐにチャットのイベント処理を追加して、 Handler

インターフェースに対応する関数を持つ構造を定義しましょう。

Rustは「伝統的な」オブジェクト指向プログラミングをサポートしていませんが、構造はクラスにほぼ類似しており、従来のOOPと同様の方法で、言語で規制されているインターフェイスを実装できます。

新しい構造を定義しましょう：

 struct WebSocketServer;

そして、 Handler

ためのHandler

を実装します。

 impl Handler for WebSocketServer { //         ,  //  Handler     :  //      . //          ,    //      mio: type Timeout = usize; type Message = (); }

次に、イベントループを実行します。

 fn main() { let mut event_loop = EventLoop::new().unwrap(); //     Handler: let mut handler = WebSocketServer; // ...        : event_loop.run(&mut handler).unwrap(); }

ここでは、最初に借用（借用）の使用に遭遇します：最後の行で&mut

に注意してください。これは、値の「所有権」を一時的に転送し、データを変更する可能性がある別の変数にリンクすることを意味します。

簡単に言えば、借用の原理は次のように想像できます（擬似コード）：

 //     "" -  owner: let mut owner = value; //          : { let mut borrow = owner; //        . //        : borrow.mutate(); //       : owner = borrow; }

上記のコードはこれと同等です：

 //     "" -  owner: let owner = value; { //     : let mut borrow = &mut owner; //     ,     . //       : borrow.mutate(); //         //     . }

スコープごとに、変数は1つの可変ボローのみを持つことができ、値の所有者でさえも、 ボローイングがスコープを離れるまで読み取りまたは変更できません。

さらに、 不変のBorrowを使用して値を借用する簡単な方法があり、読み取り専用の値を使用できます。また、変数の借用である&mut

とは異なり、読み取りに制限を設定せず、書き込みにのみ制限を設定します-スコープ内に不変の借用がある限り、値を変更して&mut

から借用することはできません。

そのような説明があなたにとって十分に明確でないように見えても大丈夫です-遅かれ早かれ直感的な理解が来るでしょう、Rustの借用はどこでも使われているので、記事を読むとより実用的な例が見つかります。

それでは、プロジェクトに戻りましょう。「 cargo run

」コマンドをcargo run

と、Cargoは必要な依存関係をすべてダウンロードし、プログラムをコンパイルします（いくつかの警告はありますが、現時点では無視できます）。

その結果、カーソルが点滅するターミナルウィンドウが表示されます。あまりおもしろい結果ではありませんが、少なくともプログラムが正しく実行されていることを示しています。これまでのところ有用なことは何もしていませんが、イベントループを正常に起動しました。この状況を修正しましょう。

プログラムを中断するには、キーボードショートカットCtrl + Cを使用します。

7 TCPサーバー

WebSocketプロトコル経由の接続を受け入れるTCPサーバーを起動するには、このために設計された構造（構造体）を使用しますTcpListener

mio::tcp

パッケージのTcpListener

。サーバーTCPソケットを作成するプロセスは非常に簡単です。特定のアドレス（IP +ポート番号）にバインドし、ソケットをリッスンして接続を受け入れます。私たちは彼をあまり離れません。

コードを見てください：

 use mio::tcp::*; use std::net::SocketAddr; ... let address = "0.0.0.0:10000".parse::<SocketAddr>().unwrap(); let server_socket = TcpListener::bind(&address).unwrap(); event_loop.register(&server_socket, Token(0), EventSet::readable(), PollOpt::edge()).unwrap();

行ごとに見てみましょう。

まず、 main.rs

モジュールのスコープに、TCPを操作するためのパッケージと、ソケットアドレスを記述するSocketAddr

構造体をインポートする必要があります。これらの行をファイルの先頭に追加します。

 use mio::tcp::*; use std::net::SocketAddr;

文字列"0.0.0.0:10000"

を解析して、アドレスを説明する構造に変換し、ソケットをこのアドレスにバインドします。

 let address = "0.0.0.0:10000".parse::<SocketAddr>().unwrap(); server_socket.bind(&address).unwrap();

コンパイラーが必要なタイプの構造を表示する方法に注意してください： server_socket.bind

はSockAddr

タイプの引数を予期するため、明示的に指定してコードを詰まらせる必要はありませんSockAddr

コンパイラーはそれを独立して決定できます。

リスニングソケットを作成して、リスニングを開始します。

 let server_socket = TcpListener::bind(&address).unwrap();

また、関数の実行結果でunwrap

を呼び出すほとんどすべての場所に気づいたかもしれません。これはRustのエラー処理のパターンであり、すぐにこのトピックに戻ります。

次に、作成したソケットをイベントループに追加しましょう。

 event_loop.register(&server_socket, Token(0), EventSet::readable(), PollOpt::edge()).unwrap();

register

呼び出しregister

より複雑です-関数は次の引数を取ります：

トークンは、ソケットの一意の識別子です。イベントがループに陥ると、それがどのソケットを指しているかを何らかの方法で理解する必要があります。この場合、 トークンはソケットとそれらが生成するイベントの間のリンクとして機能します。上記の例では、 Token(0)

トークンを接続を待機しているサーバーソケットに関連付けます。
EventSetは、サブスクライブしているイベント（ソケットへの新しいデータの到着、記録用のソケットの可用性、またはその両方）を記述します。 EventSet::readable()

サーバーソケットの場合、1つのイベントのみをサブスクライブします-新しいクライアントとの接続を確立します。
PollOptは、イベントサブスクリプション設定を設定します。 PollOpt::edge()

は、イベントがレベル（level-triggered）ではなく、エッジ（edge-triggered）でトリガーされることを意味します。

名前が電子機器から借用されている2つのアプローチの違いは、ソケットが発生したイベントを通知する瞬間にあります-たとえば、データイベントが発生したとき（つまり、 readable()

イベントにサブスクライブした場合）、レベルごとにトリガーした場合、ソケットバッファーに読み取り可能なデータがある場合は警告します。エッジの信号の場合、 新しいデータがソケットに到着した瞬間にアラートを受信します。つまり、イベントの処理中にバッファーの内容全体を読み取らなかった場合、新しいアラートが到着するまで新しいアラートを受信しません。データ。より詳細な説明（英語）はStack Overflowの回答にあります。

次に、結果のコードをコンパイルし、 cargo run

コマンドを使用してプログラムをcargo run

ます。ターミナルでは、点滅カーソル以外は何も表示されませんが、 netstat

コマンドを個別に実行すると、ソケットがポート番号10000への接続を待機していることがわかります。

 $ netstat -ln |  grep 10000
 tcp 0 0 127.0.0.1:10000 0.0.0.0:*リッスン

8接続を受け入れる

すべてのWebSocket接続は、接続の確立（いわゆるハンドシェイク ）、HTTP経由で送信される特別なシーケンスの要求と応答の確認から始まります。つまり、WebSocketの実装を進める前に、サーバーに基本プロトコルであるHTTP / 1.1を使用して通信する方法を教える必要があります。

ただし、HTTPの一部のみが必要です。WebSocketを介して接続を確立するクライアントは、ヘッダーConnection: Upgrade

およびUpgrade: websocket

を使用してリクエストを送信し、このリクエストに特定の方法で応答する必要があります。それだけです。ファイル、静的コンテンツなどの配布を備えた本格的なWebサーバーを作成する必要はありません。 -より高度で適切なツールがあります（たとえば、同じnginx）。

WebSocket接続要求ヘッダー。

しかし、HTTPの実装を開始する前に、クライアントとの接続を確立し、クライアントからのイベントをサブスクライブするコードを記述する必要があります。

基本的な実装を検討してください。

 use std::collections::HashMap; struct WebSocketServer { socket: TcpListener, clients: HashMap<Token, TcpStream>, token_counter: usize } const SERVER_TOKEN: Token = Token(0); impl Handler for WebSocketServer { type Timeout = usize; type Message = (); fn ready(&mut self, event_loop: &mut EventLoop<WebSocketServer>, token: Token, events: EventSet) { match token { SERVER_TOKEN => { let client_socket = match self.socket.accept() { Err(e) => { println!("  : {}", e); return; }, Ok(None) => panic!(" accept  'None'"), Ok(Some(sock)) => sock }; self.token_counter += 1; let new_token = Token(self.token_counter); self.clients.insert(new_token, client_socket); event_loop.register(&self.clients[&new_token], new_token, EventSet::readable(), PollOpt::edge() | PollOpt::oneshot()).unwrap(); } } } }

たくさんのコードがあったので、ステップごとに詳細に見ていきましょう。

まず、 WebSocketServer

サーバー構造に状態を追加する必要があります-サーバーソケットと接続されたクライアントのソケットを格納します。

 use std::collections::HashMap; struct WebSocketServer { socket: TcpListener, clients: HashMap<Token, TcpStream>, token_counter: usize }

クライアントソケットを格納するには、標準コレクションライブラリのHashMap

データ構造を使用しますHashMap

std::collections

は、ハッシュテーブル（辞書および連想配列とも呼ばれます）の標準実装です。キーとして、すでにおなじみのトークンを使用します。これは、接続ごとに一意である必要があります。

まず、簡単な方法でトークンを生成できます。カウンターを使用して、新しい接続ごとに1つずつ増やします。これを行うには、構造体にtoken_counter

変数が必要です。

次に、再びmio

ライブラリから便利なHandler

型をmio

ます。

 impl Handler for WebSocketServer

トレイトの実装では、コールバック関数（callback）-readyを再定義する必要があります。再定義とは、タイプにHandler

すでにダミー関数ready

と他のコールバック関数の空白が含まれていることを意味します。型で定義された実装は、もちろん有用なことは何もしないので、関心のあるイベントを処理するために独自のバージョンの関数を定義する必要があります。

 fn ready(&mut self, event_loop: &mut EventLoop<WebSocketServer>, token: Token, events: EventSet)

この関数は、ソケットが読み取りまたは書き込み（サブスクリプションに応じて）可能になるたびに呼び出され、その呼び出しパラメーターを通じて、必要なすべての情報を取得します。イベントループの構造のインスタンス、イベントソースに関連付けられたトークン（この場合、ソケット）、およびEventSet

イベントに関する情報を含むフラグのセットを含む特別な構造（読み取り用または書き込み用のそれぞれのソケットの可用性に関する通知の場合に読み取り可能）。

リスニングソケットは読み取り可能なイベントを生成します新しいクライアントが保留中の接続のキューに入る瞬間。ただし、接続を開始する前に、イベントのソースがリスニングソケットであることを確認する必要があります。これは、パターンマッチングを使用して簡単に確認できます。

 match token { SERVER_TOKEN => { ... } }

これはどういう意味ですか？この構文は、「従来の」命令型プログラミング言語match

の標準スイッチ構成を連想させますが、より多くの機能を提供します。たとえば、Javaでは、構成体はswitch

特定のタイプのセットに制限され、列挙型の数値、文字列、および列挙に対してのみ機能します。ただし、Rustでは、match

複数の値、構造などを含む、ほぼすべてのタイプの比較を行うことができます。マッチングに加えて、match

正規表現と同様の方法でサンプルの内容または部分をキャプチャすることもできます。

上記の例では、トークンをサンプルにマッピングToken(0)

します。思い出すと、トークンはリスニングソケットに接続されています。そして、コードを読むときに意図をより明確にするために、このトークンを定数として定義しました SERVER_TOKEN

：

 const SERVER_TOKEN: Token = Token(0);

したがって、式の例match

この場合には、これに相当しますmatch { Token(0) => ... }

。

サーバーソケットを処理していると確信できたので、クライアントとの接続を確立できます。

 let client_socket = match self.socket.accept() { Err(e) => { println!("  : {}", e); return; }, Ok(None) => unreachable!(), Ok(Some(sock)) => sock };

ここでも、サンプルと比較します。今回accept()

は、typeの「ラッパー」でクライアントソケットを返す関数を実行した結果をチェックしますResult<Option<TcpStream>>

。Result

これは、Rustのエラー処理の基本となる特別なタイプです。エラー、タイムアウト（タイムアウト）などの「未定義」の結果のラッパーです。

個々のケースでは、そのような結果をどう処理するかを個別に決定できますが、もちろん、すべてのエラーを正しく処理します。ここで、すでに慣れ親しんでいる関数unwrap()

が、標準の動作を提供します。エラーの場合にプログラムの実行を中断し、関数の結果をコンテナから「アンパック」しますResult

すべてが順調である場合。したがって、を使用unwrap()

することは、即時の結果と、エラーが適切なときにプログラムの実行が停止する状況にのみ関心があることを意味します。

これはいくつかの時点で許容される動作ですが、状況の組み合わせが失敗した場合、その呼び出しによってサーバーがシャットダウンされ、すべてのユーザーが切断される可能性があるため、accept()

使用するのunwrap()

は不合理です。したがって、エラーをログに出力して実行を継続します。

 Err(e) => { println!("  : {}", e); return; },

タイプOption

はResult

、値の有無を決定する「ラッパー」のようなものです。値が存在しないことはNone

; と表示され、反対の場合、値はの形式をとりますSome(value)

。おそらく推測されるように、この型は他の言語のnullまたはNone型に匹敵します。Option

すべてのnull値がローカライズされ、（Result

たとえば）使用前に必須の「アンパック」が必要なため、安全です。NullReferenceException

あなた自身がしたくない場合、「有名な」間違い。

返されたaccept()

結果をアンパックしましょう：

 Ok(None) => unreachable!(),

この場合、結果として値None

が返される状況は不可能accept()

です。クライアント（つまり、リッスンしていない）ソケットに適用されたときにこの関数を呼び出そうとした場合にのみ返されます。また、サーバーソケットを処理していると確信しているため、通常の状況ではこのコードの実行には至らないはずです。したがって、unreachable!()

エラーでプログラムの実行を中断する特別な構造を使用します。

結果をサンプルと比較し続けます。

 let client_socket = match self.socket.accept() { ... Ok(Some(sock)) => sock }

ここで最も興味深いのmatch

は、それが単なる命令ではなく、式（つまりmatch

、結果も返す）であるため、マッチングに加えて、値をキャプチャすることもできることです。したがって、これを使用して結果を変数に割り当てることができます。上記のように、型から値をアンパックResult<Option<TcpStream>>

し、変数に割り当てますclient_socket

。

トークンカウンターを増やすことを忘れずに、受信したソケットをハッシュテーブルに保存します。

 let new_token = Token(self.token_counter); self.clients.insert(new_token, client_socket); self.token_counter += 1;

最後に、接続を確立したばかりのソケットからイベントをサブスクライブする必要があります。イベントループに登録しましょう。これは、サーバーソケットの登録とまったく同じ方法で行われますが、ここではパラメーターとして別のトークン、そしてもちろん別のソケットを提供します。

 event_loop.register(&self.clients[&new_token], new_token, EventSet::readable(), PollOpt::edge() | PollOpt::oneshot()).unwrap();

あなたは、引数のセットに別の違いに気づいたことがあります。に加えてPollOpt::edge()

、我々は新しいオプションが追加されましたPollOpt::oneshot()

。イベントがトリガーされると、ループからソケットの登録を一時的に削除するように指示します。これは、サーバーコードを簡素化するのに役立ちます。このオプションがなければ、ソケットの現在の状態を手動で監視する必要があります-今すぐ書き込みが可能か、今すぐ読み取り可能かなど。代わりに、現時点で必要なオプションとサブスクリプションのセットを使用して、毎回ソケットを単純に登録します。その上、このアプローチはマルチスレッドのイベントループに役立ちますが、次回はさらに役立ちます。

そして最後に、私たちの構造がWebSocketServer

複雑なため、イベントループでサーバー登録コードを変更する必要があります。変更は非常に簡単で、主に新しい構造の初期化に関するものです。

 let mut server = WebSocketServer { token_counter: 1, //     1 clients: HashMap::new(), //   -, HashMap socket: server_socket //       }; event_loop.register(&server.socket, SERVER_TOKEN, EventSet::readable(), PollOpt::edge()).unwrap(); event_loop.run(&mut server).unwrap();

9 Parsim HTTP

プロトコルに従ってクライアントとの接続を確立したので、着信HTTPリクエストを解析し、WebSocketプロトコルへの接続を「切り替える」（アップグレードする）必要があります。

これはかなり退屈なタスクなので、すべてを手動で行うのではなく、代わりにhttp-muncher

HTTP解析ライブラリを使用して依存関係リストに追加します。このライブラリは、Node.jsのHTTPパーサー（nginxのパーサーでもあります）をHTTPに適合させます。これにより、要求をストリーミングモードで処理でき、TCP接続に非常に役立ちます。

に依存関係を追加しましょうCargo.toml

：

 [dependencies] http-muncher = "0.2.0"

ライブラリAPIを詳細に検討することはせず、すぐにパーサーの作成に進みます。

 extern crate http_muncher; use http_muncher::{Parser, ParserHandler}; struct HttpParser; impl ParserHandler for HttpParser { } struct WebSocketClient { socket: TcpStream, http_parser: Parser<HttpParser> } impl WebSocketClient { fn read(&mut self) { loop { let mut buf = [0; 2048]; match self.socket.try_read(&mut buf) { Err(e) => { println!("  : {:?}", e); return }, Ok(None) => //      . break, Ok(Some(len)) => { self.http_parser.parse(&buf[0..len]); if self.http_parser.is_upgrade() { // ... break; } } } } } fn new(socket: TcpStream) -> WebSocketClient { WebSocketClient { socket: socket, http_parser: Parser::request(HttpParser) } } }

それでもready

、構造体の関数の実装にいくつかの変更を加える必要がありますWebSocketServer

。

 match token { SERVER_TOKEN => { ... self.clients.insert(new_token, WebSocketClient::new(client_socket)); event_loop.register(&self.clients[&new_token].socket, new_token, EventSet::readable(), PollOpt::edge() | PollOpt::oneshot()).unwrap(); ... }, token => { let mut client = self.clients.get_mut(&token).unwrap(); client.read(); event_loop.reregister(&client.socket, token, EventSet::readable(), PollOpt::edge() | PollOpt::oneshot()).unwrap(); } }

行ごとに新しいコードをもう一度確認してみましょう。

まず、ライブラリをインポートし、パーサーの制御構造を追加します。

 extern crate http_muncher; use http_muncher::{Parser, ParserHandler}; struct HttpParser; impl ParserHandler for HttpParser { }

ここでは、顔の特徴の実装を追加ParserHandler

（だけでなく、いくつかの便利な機能、コールバックが含まれ、Handler

よりmio

構造の場合にWebSocketServer

）。これらのコールバックは、パーサーが有用な情報（HTTPヘッダー、リクエストコンテンツなど）を取得するとすぐに呼び出されます。しかし、今では、クライアントがHTTP接続をWebSocketプロトコルに切り替えるために特別なヘッダーのセットを送信したかどうかを調べるだけで済みます。パーサー構造にはこれに必要な機能が既に備わっているため、現時点ではコールバックを再定義せず、標準実装を残します。

ただし、詳細が1つあります。HTTPパーサーには独自の状態があるため、構造体の新しいインスタンスを作成する必要があります。HttpParser

新しい顧客ごとに。各クライアントがパーサーの状態を保存することを前提に、個々のクライアントを記述する新しい構造を作成しましょう。

 struct WebSocketClient { socket: TcpStream, http_parser: Parser<HttpParser> }

これで、クライアントソケットを同じ場所に格納できるようになったため、サーバー構造内で定義HashMap<Token, TcpStream>

を置き換えることができHashMap<Token, WebSocketClient>

ます。

さらに、クライアントの処理に関連するコードを同じ構造に移動すると便利です。すべてを1つの関数に保持するとready

、コードはすぐに「ヌードル」に変わります。それではread

、構造に別の実装を追加しましょうWebSocketClient

。

 impl WebSocketClient { fn read(&mut self) { ... } }

この関数はパラメーターを受け入れる必要はありません-構造自体の外側に必要な状態が既にあります。

これで、クライアントからのデータの読み取りを開始できます。

 loop { let mut buf = [0; 2048]; match self.socket.try_read(&mut buf) { ... } }

ここで何が起こっていますか？無限のサイクル（構築loop { ... }

）を開始し、データを書き込むバッファーに2 KBのメモリを割り当て、そこに着信データを書き込もうとします。

呼び出しtry_read

が失敗する可能性があるため、タイプによるパターンマッチングを実行しますResult

。

 match self.socket.try_read(&mut buf) { Err(e) => { println!("  : {:?}", e); return }, ... }

次に、TCPソケットバッファーに読み込むバイトが残っているかどうかを確認します。

 match self.socket.try_read(&mut buf) { ... Ok(None) => //      . break, ... }

try_read

Ok(None)

クライアントから受信した利用可能なデータをすべて読み取った場合、結果を返します。これが発生すると、無限のサイクルを中断し、新しいイベントを待ち続けます。

最後に、呼び出しtry_read

がバッファにデータを書き込んだ場合の処理を次に示します。

 match self.socket.try_read(&mut buf) { ... Ok(Some(len)) => { self.http_parser.parse(&buf[0..len]); if self.http_parser.is_upgrade() { // ... break; } } }

ここでは、受信したデータをパーサーに送信し、WebSocketモードへの接続を「切り替える」ためのリクエストの利用可能なHTTPヘッダーをすぐに確認します（より正確には、ヘッダーが必要ですConnection: Upgrade

）。

最後の改善点はnew

、クライアント構造のインスタンスの作成をより便利にするために必要な機能ですWebSocketClient

。

 fn new(socket: TcpStream) -> WebSocketClient { WebSocketClient { socket: socket, http_parser: Parser::request(HttpParser) } }

これは、いわゆる関連関数であり、多くの点で、従来のオブジェクト指向アプローチの静的メソッドに似ています。具体的new

には、関数とコンストラクタを比較できます。ここでは単にインスタンスを作成しますWebSocketClient

が、「コンストラクター」関数がなくても同じ方法で実行できることを理解する必要があります-コンストラクター関数を使用しないと、特別な必要なしにコードが頻繁に繰り返される可能性があるため、むしろ利便性の問題です。結局、DRY原則（「繰り返さない」）が発明されました。

さらにいくつかの詳細があります。キーワードは使用しないことに注意してくださいreturn

明示的に-Rustを使用すると、関数の最後の式を結果として自動的に返すことができます。

そして、この行には明確化が必要です：

 http_parser: Parser::request(HttpParser)

ここではParser

、連想関数を使用して構造の新しいインスタンスを作成しますParser::request

。引数として、以前に定義されたstructureの作成されたインスタンスを渡しますHttpParser

。

クライアントを整理したので、サーバーコードに戻り、ハンドラーでready

このような変更を行います。

 match token { SERVER_TOKEN => { ... }, token => { let mut client = self.clients.get_mut(&token).unwrap(); client.read(); event_loop.reregister(&client.socket, token, EventSet::readable(), PollOpt::edge() | PollOpt::oneshot()).unwrap(); } }

に新しい条件を追加しましたmatch

。これSERVER_TOKEN

は、クライアントソケット内のイベントに加えて、他のすべてのトークンを処理します。既存のトークンを使用すると、ハッシュテーブルからクライアント構造の対応するインスタンスへの可変リンクを借用できます。

 let mut client = self.clients.get_mut(&token).unwrap();

ここで、read

上記で定義したこのクライアントの関数を呼び出しましょう。

 client.read();

最後に、イベントループでクライアントを再登録する必要があります（ためoneshot()

）：

 event_loop.reregister(&client.socket, token, EventSet::readable(), PollOpt::edge() | PollOpt::oneshot()).unwrap();

ご覧のとおり、クライアントソケット登録手順との違いはわずかです。実際、呼び出された関数の名前をからに変更register

しreregister

、すべて同じパラメーターを渡します。

これですべてです。クライアントがWebSocketプロトコルを使用して接続を確立したいときがわかり、そのような要求にどのように応答するかを考えることができます。

10接続確認

基本的に、このような単純なヘッダーセットを送り返すことができます。

 HTTP/1.1 101 Switching Protocols Connection: Upgrade Upgrade: websocket

1つの重要な詳細がない場合：WebSocketプロトコルは、適切に構成されたヘッダーを送信することも義務付けていますSec-WebSocket-Accept

。RFCによると、特定のルールに従ってこれを行う必要があります。クライアントSec-WebSocket-Key

から送信されたヘッダーを取得して記憶し、特定の静的文字列（"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

）を追加してから、結果をSHA-1アルゴリズムでハッシュし、最後にすべてをbase64でエンコードする必要があります。

Rust標準ライブラリにはSHA-1およびbase64を操作するための関数はありませんが、必要なライブラリはすべてcrates.ioリポジトリにあるため、これらをライブラリに追加しましょうCargo.toml

。

 [dependencies] ... rustc-serialize = "0.3.15" sha1 = "0.1.1"

このライブラリにrustc-serialize

は、base64でバイナリデータをエンコードするための関数とsha1

、SHA-1でハッシュするための関数が含まれています。

応答キーを生成する関数は非常に簡単です。

 extern crate sha1; extern crate rustc_serialize; use rustc_serialize::base64::{ToBase64, STANDARD}; fn gen_key(key: &String) -> String { let mut m = sha1::Sha1::new(); let mut buf = [0u8; 20]; m.update(key.as_bytes()); m.update("258EAFA5-E914-47DA-95CA-C5AB0DC85B11".as_bytes()); m.output(&mut buf); return buf.to_base64(STANDARD); }

関数への引数としてキーを持つ文字列へのリンクを取得しgen_key

、SHA-1ハッシュの新しいインスタンスを作成し、クライアントから送信されたキーをそれに追加し、RFCで定義された定数文字列を追加し、base64でエンコードされた文字列として結果を返します。

ただし、この関数を意図した目的に使用するには、まずクライアントからヘッダーを取得する必要がありますSec-WebSocket-Key

。前のセクションからHTTPパーサーに戻りましょう。覚えているように、このタイプをParserHandler

使用すると、新しいヘッダーを受信したときに呼び出されるコールバックを再定義できます。今こそこの機会を利用する時です-対応する構造の実装を改善しましょう：

 use std::cell::RefCell; use std::rc::Rc; struct HttpParser { current_key: Option<String>, headers: Rc<RefCell<HashMap<String, String>>> } impl ParserHandler for HttpParser { fn on_header_field(&mut self, s: &[u8]) -> bool { self.current_key = Some(std::str::from_utf8(s).unwrap().to_string()); true } fn on_header_value(&mut self, s: &[u8]) -> bool { self.headers.borrow_mut() .insert(self.current_key.clone().unwrap(), std::str::from_utf8(s).unwrap().to_string()); true } fn on_headers_complete(&mut self) -> bool { false } }

このコード自体は非常に単純ですが、ここでは新しい重要な概念である共同所有に直面しています。

ご存知のように、Rustでは値の所有者は1人だけですが、ある時点で所有権を共有する必要がある場合があります。たとえば、この場合、ハッシュテーブルで特定のヘッダーを見つける必要がありますが、同時にこれらのヘッダーを記述する必要がありますパーサーで。したがって、変数headers

- WebSocketClient

との2人の所有者を取得しますParserHandler

。

この矛盾を解決するために、Rustには特別な型Rc

があります。これは、参照カウント（ガベージコレクションの一種と考えることができる）を備えたラッパーです。本質的に、コンテナの所有権を譲渡しますRc

これは、ブラックランゲージマジックの助けを借りて多くの所有者に安全に分割できますRc

。関数を使用して値を複製するだけclone()

で、コンテナがメモリを管理します。

確かに、ここには微妙な違いがありますRc

-immutableを含む値であり、コンパイラーの制限により、なんらかの影響を与えることはできません。実際、これはデータのボラティリティに関するRustルールの結果にすぎません。変数の借用は好きなだけ行うことができますが、所有者が1人しかない場合にのみ変更できます。

そして、ここでも矛盾があります-リストに新しいヘッダーを追加する必要があるのは、この変数を1か所でのみ変更することが確実であるにもかかわらず、正式にRust規則に違反しないようにするためです。このスコアについてはコンパイラーのみが同意しませんRc

。内容を変更しようとすると、コンパイルエラーが発生します。

しかし、当然のことながら、言語でこの問題を解決するには、そこにある-それは、コンテナの別のタイプを使用していますRefCell

。彼は、内部データの不安定性のメカニズムによりそれを解決します。簡単に言えば、RefCell

それは私たちが脇に、すべての検証ルールを配置することができますランタイムの代わりに静的にそれらをチェックするので- （ランタイム）コンパイル時。したがって、ヘッダーを2つのコンテナーに同時にラップする必要がありますRc<RefCell<...>>

（もちろん、心の準備ができていない場合はかなり恐ろしく見えます）。

ハンドラーから次の行を見てみましょうHttpParser

。

 self.headers.borrow_mut() .insert(self.current_key.clone().unwrap(), ...

このような設計は全体として変数ボローイング&mut

に対応しますが、ボローイングの数を制限するすべてのチェックはプログラム実行中に動的に実行されるため、コンパイラーではなく、私たちがこれを注意深く監視する必要があります。そうしないと、ランタイムエラーが発生する可能性があります構造体は

変数の直接の所有者にheaders

なるWebSocketClient

ため、新しいプロパティを追加して、新しいコンストラクター関数を作成しましょう。

 //   RefCell  Rc    use std::cell::RefCell; use std::rc::Rc; ... struct WebSocketClient { socket: TcpStream, http_parser: Parser<HttpParser>, //      WebSocketClient: headers: Rc<RefCell<HashMap<String, String>>> } impl WebSocketClient { fn new(socket: TcpStream) -> WebSocketClient { let headers = Rc::new(RefCell::new(HashMap::new())); WebSocketClient { socket: socket, //      : headers: headers.clone(), http_parser: Parser::request(HttpParser { current_key: None, // ...     : headers: headers.clone() }) } } ... }

これでWebSocketClient

、解析済みヘッダーにアクセスできるようになったため、それらの中から興味のあるヘッダーを見つけることができます- Sec-WebSocket-Key

。クライアントキーがある場合、応答をコンパイルする手順は問題になりません。文字列を分割して収集し、クライアントソケットに書き込むだけです。

ただし、非ブロッキングソケットにデータを送信することはできないため、最初にイベントループに問い合わせて、記録用のソケットの可用性を通知する必要があります。それを簡単に-あなたはフラグのセットを変更する必要がEventSet

でEventSet::writable()

ソケットの再登録の時。

この行を覚えていますか？

 event_loop.reregister(&client.socket, token, EventSet::readable(), PollOpt::edge() | PollOpt::oneshot()).unwrap();

興味のある一連のイベントをクライアント状態に保存できます-構造を変更しますWebSocketClient

：

 struct WebSocketClient { socket: TcpStream, http_parser: Parser<HttpParser>, headers: Rc<RefCell<HashMap<String, String>>>, //   , `interest`: interest: EventSet }

それに応じて、再登録手順を適宜変更します。

 event_loop.reregister(&client.socket, token, client.interest, //    `EventSet`    PollOpt::edge() | PollOpt::oneshot()).unwrap();

interest

適切な場所で値を変更するだけです。このプロセスを簡素化するために、接続状態を使用して形式化しましょう。

 #[derive(PartialEq)] enum ClientState { AwaitingHandshake, HandshakeResponse, Connected }

ここでは、サーバーに接続されたクライアントのすべての可能な状態の列挙を定義します。最初の状態AwaitingHandshake

は、新しいクライアントがHTTP経由で接続することを期待していることを意味します。HandshakeResponse

HTTPを介してクライアントに応答するときの状態を意味します。そして最後に、Connected

クライアントとの接続を正常に確立し、WebSocketプロトコルを使用して通信するときの状態。

状態変数をクライアント構造に追加します。

 struct WebSocketClient { socket: TcpStream, http_parser: Parser<HttpParser>, headers: Rc<RefCell<HashMap<String, String>>>, interest: EventSet, //   : state: ClientState }

そして、新しい変数の初期値をコンストラクターに追加します。

 impl WebSocketClient { fn new(socket: TcpStream) -> WebSocketClient { let headers = Rc::new(RefCell::new(HashMap::new())); WebSocketClient { socket: socket, ... // Initial events that interest us interest: EventSet::readable(), // Initial state state: ClientState::AwaitingHandshake } } }

これで、関数の状態を変更できますread

。これらの行を覚えていますか？

 match self.socket.try_read(&mut buf) { ... Ok(Some(len)) => { if self.http_parser.is_upgrade() { // ... break; } } }

条件ブロックのスタブをis_upgrade()

、接続ステータスを変更するためのコードに変更します。

 if self.http_parser.is_upgrade() { //     HandshakeResponse self.state = ClientState::HandshakeResponse; //       Writable // (..    ): self.interest.remove(EventSet::readable()); self.interest.insert(EventSet::writable()); break; }

対象のフラグのセットをに変更した後Writable

、接続を確立するために応答を送信するために必要なコードを追加します。構造体の実装で

関数を変更します。ソケット自体への応答の書き込み手順は単純で（実際には読み取り手順と変わらない）、1つのタイプのイベントを他のタイプから分離するだけで済みます。ready

WebSocketServer

 fn ready(&mut self, event_loop: &mut EventLoop<WebSocketServer>, token: Token, events: EventSet) { //        ? if events.is_readable() { // Move all read handling code here match token { SERVER_TOKEN => { ... }, ... } ... } //        : if events.is_writable() { let mut client = self.clients.get_mut(&token).unwrap(); client.write(); event_loop.reregister(&client.socket, token, client.interest, PollOpt::edge() | PollOpt::oneshot()).unwrap(); } }

少しだけ残っています-部分的に収集して応答行を送信する必要があります：

 use std::fmt; ... impl WebSocketClient { fn write(&mut self) { //  -    Rc<RefCell<...>>: let headers = self.headers.borrow(); //           : let response_key = gen_key(&headers.get("Sec-WebSocket-Key").unwrap()); //       . //         (printf  , format  Python,  ..), //   Rust    -        // ,         ""  //  .           . let response = fmt::format(format_args!("HTTP/1.1 101 Switching Protocols\r\n\ Connection: Upgrade\r\n\ Sec-WebSocket-Accept: {}\r\n\ Upgrade: websocket\r\n\r\n", response_key)); //    : self.socket.try_write(response.as_bytes()).unwrap(); //    : self.state = ClientState::Connected; //         `readable()` ( ): self.interest.remove(EventSet::writable()); self.interest.insert(EventSet::readable()); } }

サーバーに接続してみましょう。お気に入りのブラウザーで開発コンソールを開き（たとえば、F12を押して）、次のコードを入力します。

 ws = new WebSocket('ws://127.0.0.1:10000'); if (ws.readyState == WebSocket.OPEN) { console.log('Connection is successful'); }

すべてが機能しているようです-サーバーに接続しています！

おわりに

Rust言語の可能性と珍しい概念を巡る魅力的な旅は終わりましたが、最初に触れただけです-一連の記事が続きます（もちろん、続編は長くて退屈です！:)）。他の多くの興味深い問題を考慮する必要があります：セキュアTLS接続、マルチスレッドイベントサイクル、負荷テストと最適化、そしてもちろん、最も重要なこと-WebSocketプロトコルの実装を終了し、チャットアプリケーション自体を記述する必要があります。

しかし、アプリケーションに到達する前に、ライブラリコードをアプリケーションコードから少しリファクタリングして分離する必要があります。ほとんどの場合、crates.ioでライブラリを公開することも検討します。

現在のコードはすべてGithubで入手できます。、リポジトリをフォークして、リポジトリ内の何かを変更してみてください。

記事の以下の部分の外観をフォローするには、Twitterでフォローしてください。

じゃあね！

注釈

[1]本質的に使用していること、錆に注意すべきでスマートポインタを、言語のレベルに-借入のアイデアは、種類と非常によく似ているunique_ptr

とshared_ptr

C ++から。

[2]たとえば、NASAのJet Propulsion LaboratoryのNASコーディング標準および自動車業界標準のMISRA Cは、一般に、を介した動的メモリ割り当ての使用を禁止していますmalloc()

。代わりに、スタック上のローカル変数の割り当てと事前に割り当てられたメモリの使用を想定しています。

[3]単純なガベージコレクションアルゴリズムは非常に使いやすいですが、マルチスレッドアセンブリなどのより複雑なオプションでは、かなりの実装作業が必要になる場合があります。たとえば、Go言語では、マルチスレッドガベージコレクションはバージョン1.5にのみ登場し、バージョン1.5は最初のリリースからほぼ3年後に登場しました。

[4]一般的に言えば、多くの関数の実装にmalloc()

もメモリの断片化free()

による同じ問題があります。

[5]「Graydon Hoar [...]は2006年にRustと呼ばれる新しいプログラミング言語で作業を開始しました」-InfoQ：「Rustのインタビュー」

[6]マニュアルページでpthread_create(3)

は、32ビットLinuxシステムで2 MBについて説明しています。

[7] epollを他のシステムAPIと比較するには、出版物「epoll、select、およびポーリングイベントメカニズムの比較と評価、ウォータールー大学、2004

[8]「Kqueue：一般的でスケーラブルなイベント通知機能」（

9）「内部からのNGINX：パフォーマンスとスケーリング。」

私は助けに感謝します：イラストと校正のための

ポダスト。

下書きを読んで校正するためのVgaCich。

錆の詳細：スケーラブルなチャットをゼロから作成する、パート1

パート1：WebSocketを実装します。 はじめに