# IO、ポートとネットワーキング

> Source: https://www.ymotongpoo.com/works/beam-book-ja/understanding_erts/io/


<a id="CH-IO"></a>

Erlangの内部では、あらゆる通信が非同期のシグナルによって行われます。Erlangノードと外の世界との通信は*ポート*を通して行われます。ポートはErlangプロセスと外部リソースの間のインターフェースです。初期のバージョンのErlangでは、ポートはプロセスとよく似た振る舞いをし、シグナルの送受信によって通信していました。今でもこの方法でポートと通信できますが、ポートと直接やり取りするためのBIFも数多く用意されています。

この章では、ポートがあらゆるIOの共通インターフェースとしてどう使われているか、ポートが外の世界とどう通信するか、そしてErlangプロセスがポートとどう通信するかを見ていきます。ただし、その前にまず標準入出力が上位のレベルでどう動いているかを見ておきます。

<a id="_standard_io"></a>

## 標準入出力

Erlangにおける標準入出力の仕組みを理解しておくと、デバッグや外部プログラムとのやり取りに役立ちます。この節では、I/Oプロトコル、グループリーダー、`erlang:display`や`io:format`の使い方、標準入出力をリダイレクトする方法を扱います。

<a id="_io_protocol"></a>

### I/Oプロトコル

Erlangの I/Oプロトコルは、プロセスとI/Oデバイスの間の通信を扱います。このプロトコルは、端末やファイル、外部プログラムといったデバイスとの間でデータをどう送受信するかを定めています。プロトコルには、読み取り、書き込み、フォーマット、I/O制御操作を扱うためのコマンドが含まれます。これらのコマンドは非同期に実行され、Erlangの並行性モデルを保っています。より詳しい情報は、公式ドキュメントの[erlang.org:io_protocol](https://www.erlang.org/doc/apps/stdlib/io_protocol.html)を参照してください。

I/Oプロトコルは、I/Oリクエストの処理を担当するグループリーダーとの通信に使われます。グループリーダーは、他のプロセスからI/OリクエストAを受け取り、それをI/Oサーバーへ転送するプロセスです。I/Oサーバーはリクエストを実行し、その応答をグループリーダーへ送り返す役割を持ち、グループリーダーはそれをリクエスト元のプロセスへ転送します。

グループリーダーについては後で説明するとして、まずは実際のI/Oプロトコルを見ていきましょう。このプロトコルには次のようなメッセージがあります。

<a id="_basic_messages"></a>

#### 基本メッセージ

- `{io_request, From, ReplyAs, Request}`：
  - `From`：クライアントプロセスの`pid()`。
  - `ReplyAs`：応答をリクエストに対応付けるための識別子。
  - `Request`：I/Oリクエストの内容。

- `{io_reply, ReplyAs, Reply}`：
  - `ReplyAs`：元のリクエストに対応する識別子。
  - `Reply`：I/Oリクエストへの応答。

<a id="_output_requests"></a>

#### 出力リクエスト

- `{put_chars, Encoding, Characters}`：
  - `Encoding`：`unicode`または`latin1`。
  - `Characters`：書き込むデータ。

- `{put_chars, Encoding, Module, Function, Args}`：
  - `Module, Function, Args`：データを生成する関数。

<a id="_input_requests"></a>

#### 入力リクエスト

- `{get_until, Encoding, Prompt, Module, Function, ExtraArgs}`：
  - `Encoding`：`unicode`または`latin1`。
  - `Prompt`：プロンプトとして出力するデータ。
  - `Module, Function, ExtraArgs`：十分な量のデータが読み込まれたかどうかを判定する関数。

- `{get_chars, Encoding, Prompt, N}`：
  - `Encoding`：`unicode`または`latin1`。
  - `Prompt`：プロンプトとして出力するデータ。
  - `N`：読み込む文字数。

- `{get_line, Encoding, Prompt}`：
  - `Encoding`：`unicode`または`latin1`。
  - `Prompt`：プロンプトとして出力するデータ。

<a id="_server_modes"></a>

#### サーバーモード

- `{setopts, Opts}`：
  - `Opts`：I/Oサーバーに対するオプションのリスト。

- `getopts`：
  - I/Oサーバーに現在のオプションを問い合わせます。

<a id="_multi_request_and_optional_messages"></a>

#### 複数リクエストと省略可能なメッセージ

- `{requests, Requests}`：
  - `Requests`：順に実行される有効な`io_request`タプルのリスト。

- `{get_geometry, Geometry}`：
  - `Geometry`：行数または列数を問い合わせます（省略可能）。

<a id="_unimplemented_request_handling"></a>

#### 未実装リクエストの扱い

I/Oサーバーが認識できないリクエストを受け取った場合は、`{error, request}`を返すべきです。

<a id="_example_of_a_custom_io_server"></a>

#### 自作I/Oサーバーの例

ここでは、データをメモリ上に保持するI/Oサーバーの簡単な例を示します。

``` erlang
-module(custom_io_server).
-export([start_link/0, stop/1, init/0, loop/1, handle_request/2]).

-record(state, {buffer = <<>>, pos = 0}).

start_link() ->
    {ok, spawn_link(?MODULE, init, [])}.

init() ->
    ?MODULE:loop(#state{}).

stop(Pid) ->
    Pid ! {io_request, self(), Pid, stop},
    receive
        {io_reply, _, {ok, State}} ->
            {ok, State#state.buffer};
        Other ->
            {error, Other}
    end.

loop(State) ->
    receive
        {io_request, From, ReplyAs, Request} ->
            case handle_request(Request, State) of
                {ok, Reply, NewState} ->
                    From ! {io_reply, ReplyAs, Reply},
                    loop(NewState);
                {stop, Reply, _NewState} ->
                    From ! {io_reply, ReplyAs, Reply},
                    exit(normal);
                {error, Reply, NewState} ->
                    From ! {io_reply, ReplyAs, {error, Reply}},
                    loop(NewState)
            end
    end.

handle_request({put_chars, _Encoding, Chars}, State) ->
    Buffer = State#state.buffer,
    NewBuffer = <<Buffer/binary, Chars/binary>>,
    {ok, ok, State#state{buffer = NewBuffer}};
handle_request({get_chars, _Encoding, _Prompt, N}, State) ->
    Part = binary:part(State#state.buffer, State#state.pos, N),
    {ok, Part, State#state{pos = State#state.pos + N}};
handle_request({get_line, _Encoding, _Prompt}, State) ->
    case binary:split(State#state.buffer, <<$\n>>, [global]) of
        [Line|_Rest] ->
            {ok, <<Line/binary, $\n>>, State};
        _ ->
            {ok, State#state.buffer, State}
    end;
handle_request(getopts, State) ->
    {ok, [], State};
handle_request({setopts, _Opts}, State) ->
    {ok, ok, State};
handle_request(stop, State) ->
    {stop, {ok, State}, State};
handle_request(_Other, State) ->
    {error, {error, request}, State}.
```

このメモリストアは、たとえばファイルインターフェースを介したI/Oデバイスとして使えます。

``` erlang
-module(file_client).
-export([open/0, close/1, write/2, read/2, read_line/1]).

open() ->
    {ok, Pid} = custom_io_server:start_link(),
    {ok, Pid}.

close(Device) ->
    custom_io_server:stop(Device).

write(Device, Data) ->
    file:write(Device, Data).

read(Device, Length) ->
    file:read(Device, Length).

read_line(Device) ->
    file:read_line(Device).
```

これで、`file_client`インターフェースを通してこのメモリストアを使えるようになりました。

``` erlang
shell V14.2.1 (press Ctrl+G to abort, type help(). for help)
1> {ok, Pid} = file_client:open().
{ok,<0.219.0>}
2> file_client:write(Pid, "Hello, world!\n").
ok
3> R = file_client:close(Pid).
{ok,<<"Hello, world!\n">>}
4>
```

<a id="_group_leader"></a>

### グループリーダー

グループリーダーを使うと、I/Oを適切な宛先へリダイレクトできます。この性質は子プロセスに引き継がれ、連鎖を作ります。デフォルトでは、Erlangノードには`user`という名前のグループリーダーがあり、これが標準入出力チャネルとの通信を管理しています。すべての入出力リクエストはこのプロセスを経由します。

Erlangで起動された各シェルは、それ自体が独立したグループリーダーになります。つまり、そのシェルから実行された関数は、すべてのI/Oデータをそのシェルプロセスへ送ることになります。^Gでシェルを切り替えて（たとえば`c <番号>`のように）別のシェルを選ぶと、シェル管理用の特別なプロセスがI/Oのやり取りを正しいシェルへ振り向けます。

分散システムでは、スレーブノードやリモートシェルはグループリーダーを別ノード上のPIDに設定し、子孫プロセスからのI/Oデータが正しくルーティングされるようにします。

各OTPアプリケーションには、グループリーダーとして振る舞うアプリケーションマスタープロセスがあります。これには主に2つの用途があります。

1. プロセスが`application:get_env(Var)`を使って自分のアプリケーションの環境設定にアクセスできるようにすること。
2. アプリケーションのシャットダウン時に、アプリケーションマスターが同じグループリーダーを持つすべてのプロセスを走査して終了させ、事実上アプリケーションのプロセスをガベージコレクションすること。

グループリーダーは、`common_test`や`eunit`によるテスト中にI/Oを捕捉する用途にも使われ、対話的なシェルでもI/Oを管理するためにグループリーダーが設定されます。

<a id="_group_leader_functions"></a>

#### グループリーダー関数

- `group_leader() -> pid()`：プロセスのグループリーダーのPIDを返します。
- `group_leader(GroupLeader, Pid) -> true`：`Pid`のグループリーダーを`GroupLeader`に設定します。

スーパービジョンツリーを持つアプリケーションでは、プロセスのグループリーダーが変更されることは通常ありません。OTPは、グループリーダーがアプリケーションマスターであることを前提にしているためです。

`group_leader/2`関数は、`group_leader`シグナルを使ってプロセスのグループリーダーを設定します。このシグナルは対象のプロセスへ送られ、そのプロセスは自分のグループリーダーを指定されたPIDに設定します。グループリーダーには任意のプロセスを指定できますが、多くの場合はシェルプロセスかアプリケーションマスターです。

<a id="_example_usage"></a>

#### 使用例

``` erlang
1> group_leader().
<0.24.0>

2> self().
<0.42.0>

3> group_leader(self(), <0.43.0>).
true
```

グループリーダーの仕組みと、BIFである`display`と関数`io:format`の違いを理解しておくと、Erlangにおける基本的なI/Oの管理に役立ちます。

- `erlang:display/1`：ErlangのI/Oシステムを経由せずに、標準出力へ直接書き込むBIFです。
- `io:format/1,2`：I/Oリクエストをグループリーダーへ送ります。`rpc:call/4`経由で実行された場合、出力は呼び出し元プロセスの標準出力へ向かいます。

<a id="_redirecting_standard_io_at_startup_detached_mode"></a>

### 起動時の標準入出力のリダイレクト（デタッチモード）

Erlangでは、起動時に標準入出力（stdin、stdout、stderr）をリダイレクトできます。とくにデタッチモードで動かす場合は、入出力の向き先を自分で制御できるようになります。これは、出力をファイルへログとして残したい場合や、別のソースから入力を受け取りたい場合など、本番環境で特に役立ちます。

<a id="_detached_mode"></a>

#### デタッチモード

Erlangのデタッチモードは、`-detached`フラグを使うと有効にできます。このフラグを付けると、Erlangランタイムシステムは接続されたコンソールを持たないバックグラウンドプロセスとして起動します。デタッチモードでErlangノードを起動する例は次のとおりです。

``` sh
erl -sname mynode -setcookie mycookie -detached
```

デタッチモードで動かす場合、接続されたコンソールが存在しないため、標準入出力は手動でリダイレクトする必要があります。これは、Erlangランタイムに対してリダイレクト用のオプションを指定することで行えます。

<a id="_redirecting_standard_output_and_error"></a>

#### 標準出力と標準エラーのリダイレクト

標準出力と標準エラーをファイルへリダイレクトするには、シェルのリダイレクト機能かErlangの組み込みオプションを使います。stdoutとstderrをそれぞれ別のログファイルへリダイレクトする例を示します。

``` sh
erl -sname mynode -setcookie mycookie -detached > mynode_stdout.log 2> mynode_stderr.log
```

このコマンドは、デタッチモードでErlangノードを起動し、標準出力を`mynode_stdout.log`へ、標準エラーを`mynode_stderr.log`へリダイレクトします。

これは、Erlangのスクリプトや起動設定の中で構成することもできます。

``` erlang
init() ->
    %% stdoutとstderrをリダイレクトする
    file:redirect(standard_output, "mynode_stdout.log"),
    file:redirect(standard_error, "mynode_stderr.log").
```

<a id="_redirecting_standard_input"></a>

#### 標準入力のリダイレクト

標準入力をリダイレクトするには、Erlangノードの起動時に入力ソースを指定します。たとえば、ファイルを入力ソースとして使うことができます。

``` sh
erl -sname mynode -setcookie mycookie -detached < input_commands.txt
```

<a id="_standard_input_and_output_summary"></a>

### 標準入出力のまとめ

標準出力（`stdout`）はグループリーダーによって管理され、通常はコンソールか指定されたログファイルへ向けられます。`io:format/1`や`io:put_chars/2`といった関数は出力をグループリーダーへ送り、グループリーダーがそれを指定された出力デバイスやファイルへ書き込みます。標準エラー（`stderr`）も同様にグループリーダーによって管理され、ログファイルなど他の出力先へリダイレクトすることもできます。標準入力（`stdin`）は、プロセスがグループリーダーから読み取ります。デタッチモードでは、入力をファイルなど別の入力ソースからリダイレクトできます。

I/Oプロトコルを使って自分だけのI/Oサーバーを実装し、それをファイルディスクリプタとして使ったり、プロセスのグループリーダーとして設定してそのサーバー経由でI/Oをリダイレクトしたりすることもできます。

<a id="_ports_2"></a>

## ポート

ポートは、Erlangプロセスと、Erlangプロセスではないあらゆるものとの間をつなぐ、プロセスのようなインターフェースです。プログラマは、世界のあらゆるものがErlangプロセスのように振る舞い、メッセージパッシングによって通信していると、かなりの程度まで見なして扱うことができます。

すべてのポートにはオーナーがいます（詳しくは後述します）が、そのポートを知っているすべてのプロセスはそのポートへメッセージを送ることができます。次の図では、あるプロセスがポートとどう通信するか、そしてそのポートがErlangノードの外の世界とどう通信するかを示しています。

<a id="port_communication"></a>

![ポート通信](20260830-port-communication.png)
*図: ポート通信。実線の両矢印は送受信、片矢印は送信のみを表す。プロセスP1はPort1のオーナーとして送受信を行い、プロセスP2は送信のみを行う。Port1はファイルとの間で読み書きを行う。*

プロセスP1はファイルに対してポート（Port1）を開いており、そのポートのオーナーとしてポートからメッセージを受け取ることができます。プロセスP2もこのポートへのハンドルを持っていて、ポートへメッセージを送ることができます。プロセスとポートはErlangノードの中に存在します。ファイルは、Erlangノードの外にあるファイルシステムおよびオペレーティングシステムの中に存在します。

ポートのオーナーが死ぬか終了させられると、そのポートも終了します。ポートが終了するときには、外部のリソースもすべて後始末されるべきです。これはErlangに標準で付属するすべてのポートについて当てはまることで、自分でポートを実装する場合も、この後始末を行うようにする必要があります。

<a id="_different_types_of_ports"></a>

### ポートの種類

ポートには、ファイルディスクリプタ、外部プログラム、ドライバという3つの異なる分類があります。ファイルディスクリプタポートを使うと、プロセスはすでに開かれているファイルディスクリプタにアクセスできます。外部プログラムへのポートは、その外部プログラムを別のOSプロセスとして起動します。ドライバポートを使うには、Erlangノードにドライバがロードされている必要があります。

すべてのポートは、`erlang:open_port(PortName, PortSettings)`の呼び出しによって作られます。

ファイルディスクリプタポートは、`PortName`として`{fd, In, Out}`を渡すことで開かれます。このクラスのポートは、古いシェルのような一部の内部ERTSサーバーで使われています。効率があまり良くないと考えられており、そのためあまり使われていません。また、ファイルディスクリプタはOS上で開かれているファイルディスクリプタを表す非負の整数です。ファイルディスクリプタをErlangのI/Oサーバーにすることはできません。

外部プログラムポートは、ErlangノードのネイティブなOS上で任意のプログラムを実行するために使えます。外部プログラムポートを開くには、その外部プログラムの名前とともに引数`{spawn, Command}`または`{spawn_executable, FileName}`を渡します。これは、他のプログラミング言語で書かれたコードとやり取りする、もっとも簡単で、もっとも安全な方法の1つです。外部プログラムは自分自身のOSプロセスとして実行されるため、それがクラッシュしてもErlangノードを道連れにすることはありません（もちろん、CPUやメモリを使い尽くすなどしてOS全体を巻き込むことはあり得ますが、リンクインドライバやNIFに比べればはるかに安全です）。

ドライバポートを使うには、ERTSにドライバプログラムがロードされている必要があります。このようなポートは、`{spawn, Command}`または`{spawn_driver, Command}`のどちらかで起動します。独自のリンクインドライバを書くことは、たとえば利用したいCライブラリのコードに対するインターフェースを効率よく実装する方法になり得ます。ただし、リンクインドライバはErlangノードと同じOSプロセス内で実行されるため、ドライバ側のクラッシュはノード全体を巻き込むことに注意してください。Erlangドライバの書き方に関する詳細は[Cの章](../c/#CH-C)にあります。

Erlang/OTPには、あらかじめ定義されたポートの種類を実装する数多くのポートドライバが付属しています。すべてのプラットフォームで使える共通のドライバとしては、`tcp_inet`、`udp_inet`、`sctp_inet`、`efile`、`zlib_drv`、`ram_file_drv`、`binary_filer`、`tty_sl`があります。これらのドライバは、たとえばErlangにおけるファイル操作やソケットの実装に使われています。Windowsでは、レジストリにアクセスするための`registry_drv`というドライバもあります。また、ほとんどのプラットフォームには、独自ドライバを実装する際の見本として使える`multi_drv`や`sig_drv`といったサンプルドライバがあります。

<a id="entities_on_node"></a>

![Erlangノード上のエンティティ](20260830-entities-on-node.png)
*図: Erlangノード上のエンティティ。矢印はない。プロセスとポートを対比し、ポートはさらにFD・OS・ドライバに分類される。各分類の下の行は実例を示す（FDの例は`(heart)`、OSの例は起動コマンド`(ls)`、ドライバの例は`tcp_inet`や`efile`）。*

ポートとの間で送受信されるデータはバイトストリームです。ポートを開く際の`PortSettings`でパケットサイズを指定できます。R16以降、ポートは真に非同期な通信をサポートしており、効率と性能が向上しています。

ポートは、標準入出力の代わりやポーリングの代わりとして使うこともできます。これは、外部プログラムやデバイスとやり取りする必要がある場合に便利です。ファイルディスクリプタへポートを開くことで、そのファイルへの読み書きができます。同様に、外部プログラムへポートを開いて、ポートインターフェースを通してそのプログラムとやり取りすることもできます。

<a id="_ports_to_file_descriptors"></a>

#### ファイルディスクリプタへのポート

Erlangのファイルディスクリプタポートは、すでに開かれているファイルディスクリプタとやり取りするためのインターフェースを提供します。効率面の懸念からあまり一般的には使われていませんが、外部リソースへの手軽なインターフェースとして使えます。

ファイルディスクリプタポートを作るには、`open_port/2`関数に、`PortName`として`{fd, In, Out}`タプルを渡します。ここで`In`と`Out`は、それぞれ入力と出力のファイルディスクリプタです。

悪い例です。

``` erlang
Port = open_port({fd, 0, 1}, []).
```

これは、標準入力（ファイルディスクリプタ0）から読み込み、標準出力（ファイルディスクリプタ1）へ書き込むポートを開きます。この例は、自分のErlangシェルからIOを奪ってしまうので、絶対に試さないでください。

それでも試してしまいましたね。今すぐシェルを再起動する必要があります。

ファイルディスクリプタポートは、`open_port/2`関数を使って実装されており、これがポートオブジェクトを作ります。このポートオブジェクトが、Erlangプロセスとファイルディスクリプタの間の通信を扱います。

内部的には、`{fd, In, Out}`を指定して`open_port/2`が呼ばれると、Erlangランタイムシステムは指定されたファイルディスクリプタとやり取りするために必要な通信チャネルを設定します。ポートのオーナープロセスは、そのポートに対してメッセージを送受信でき、ポートはそれを受けてファイルディスクリプタとやり取りします。

<a id="_ports_to_spawned_os_processes"></a>

#### 起動したOSプロセスへのポート

起動したOSプロセスへのポートを作るには、`open_port/2`関数に、`PortName`として`{spawn, Command}`または`{spawn_executable, FileName}`タプルを渡します。この方法により、Erlangプロセスは外部プログラムを別のOSプロセスとして起動し、それとやり取りできます。

ポートとやり取りするための主なコマンドには、次のものがあります。

- `{command, Data}`：外部プログラムへ`Data`を送ります。
- `{control, Operation, Data}`：外部プログラムへ制御コマンドを送ります。
- `{exit_status, Status}`：外部プログラムの終了ステータスを受け取ります。

外部プログラムをポートとして起動する例は[Cの章](../c/#CH-C)を参照してください。公式ドキュメントの[erlang.org:c_port](https://www.erlang.org/doc/system/c_port.html#content)も参考になります。

<a id="_ports_to_linked_in_drivers"></a>

#### リンクインドライバへのポート

Erlangのリンクインドライバは、`open_port/2`関数に、`PortName`として`{spawn_driver, Command}`タプルを渡すことで作られます。この方法では、コマンドの最初のトークンがロード済みドライバの名前である必要があります。

``` erlang
Port = open_port({spawn_driver, "my_driver"}, []).
```

リンクインドライバポートとやり取りするためのコマンドには、通常次のものが含まれます。

- `{command, Data}`：ドライバへ`Data`を送ります。
- `{control, Operation, Data}`：ドライバへ制御コマンドを送ります。例は次のとおりです。

``` erlang
Port ! {self(), {command, <<"Hello, Driver!\n">>}}.
```

リンクインドライバをポートとして実装・起動する例は[Cの章](../c/#CH-C)を参照してください。公式ドキュメントの[erlang.org:c_portdriver](https://www.erlang.org/doc/system/c_portdriver.html#content)も参考になります。

<a id="_flow_control_in_erlang_ports"></a>

### Erlangポートにおけるフロー制御

ポートは、バックプレッシャーを管理し効率的なリソース利用を確実にするためのフロー制御の仕組みを備えています。その主な仕組みの1つがビジーポート機能で、これはポートが同時に大量の操作にさらされて処理しきれなくなるのを防ぎます。ポートがビジー状態になると、既存のデータを処理し終えるまでこれ以上データを受け付けられないことをErlang VMへ知らせることができます。

ポート内部のバッファが指定されたハイウォーターマークを超えると、そのポートはビジー状態に入ります。この状態になると、バッファ済みのデータを処理できるようになるまで新しいデータの送信を止めるよう、VMへ信号を送ります。

ビジー状態のポートへデータを送ろうとするプロセスは、ポートがビジー状態を抜けるまで一時停止されます。これにより、データの欠落を防ぎつつ、受信済みのすべてのデータをポートが効率よく処理できるようにします。

ポートが十分な量のデータを処理して、指定されたロウウォーターマークを下回ると、ビジー状態を抜けます。一時停止していたプロセスは、そこでデータの送信を再開できるようになります。

ポートとの間のシグナルを非同期にスケジューリングすることで、Erlangはデータを送信するプロセスがブロックされることなく実行を続けられるようにし、システムの並列性と応答性を高めています。

つまり、これはErlangの送信操作が常に非同期であるとは限らないということでもあります。ポートがビジーであれば、送信操作はそのポートがビジーでなくなるまでブロックします。同じポートへデータを送るプロセスが多数ある場合、これは問題になります。解決策は、バックプレッシャーを扱い、送信操作が常に非同期になるようにしてくれるポートサーバーを使うことです。

公式のポートドライバの例[erlang.org:c-driver](https://www.erlang.org/doc/system/c_portdriver.html#c-driver)にもとづいて、実際に例を作ってみましょう。

まずは、1を足す、あるいは2倍にするという、元のシンプルなC関数を使います。

``` c
/* example.c */

int foo(int x)
{
    return x + 1;
}

int bar(int y)
{
    return y * 2;
}
```

そして、これを少し手直ししたポートドライバです。各メッセージにIDを追加し、そのIDを結果とともに返すようにしています。また、メッセージのIDが14未満で呼び出し先が`bar`関数の場合に、しばらくスリープしてビジーポート状態を設定することで、ビジーポートをシミュレートする関数も追加しています。

``` c
/* port_driver.c を元にしている
   https://www.erlang.org/doc/system/c_portdriver.html

*/

#include "erl_driver.h"
#include <stdio.h>
#include <unistd.h> // sleep関数のためにインクルード

int foo(int x);
int bar(int y);

typedef struct
{
    ErlDrvPort port;
} example_data;

static ErlDrvData bp_drv_start(ErlDrvPort port, char *buff)
{
    example_data *d = (example_data *)driver_alloc(sizeof(example_data));
    d->port = port;
    return (ErlDrvData)d;
}

static void bp_drv_stop(ErlDrvData handle)
{
    driver_free((char *)handle);
}

static void bp_drv_output(ErlDrvData handle, char *buff,
                          ErlDrvSizeT bufflen)
{
    example_data *d = (example_data *)handle;
    char fn = buff[0], arg = buff[1], id = buff[2];
    static char res[2];

    if (fn == 1)
    {
        res[0] = foo(arg);
    }
    else if (fn == 2)
    {
        res[0] = bar(arg);
        if (id > 14)
        {
            // ポートが空いたことを知らせる
            set_busy_port(d->port, 0);
        }
        else
        {
            // ポートがビジーであることを知らせる
            // この例では本質的な部分ではない
            // ただし、複数のプロセスが同時にこのポートを
            // 使おうとした場合は、ポートがビジーであることを
            // 知らせる必要がある
            // そうすると foo 関数さえもブロックすることになる
            set_busy_port(d->port, 1);
            // 処理の遅延をシミュレートする
            sleep(1);
            set_busy_port(d->port, 0);
        }
    }
    res[1] = id;
    driver_output(d->port, res, 2);
}

ErlDrvEntry bp_driver_entry = {
    NULL,                           /* F_PTR init, ドライバのロード時に呼ばれる */
    bp_drv_start,                   /* L_PTR start, ポートが開かれたときに呼ばれる */
    bp_drv_stop,                    /* F_PTR stop, ポートが閉じられたときに呼ばれる */
    bp_drv_output,                  /* F_PTR output, Erlangがデータを送ったときに呼ばれる */
    NULL,                           /* F_PTR ready_input, 入力ディスクリプタの準備ができたときに呼ばれる */
    NULL,                           /* F_PTR ready_output, 出力ディスクリプタの準備ができたときに呼ばれる */
    "busy_port_drv",                /* char *driver_name, open_port に渡す引数 */
    NULL,                           /* F_PTR finish, アンロード時に呼ばれる */
    NULL,                           /* void *handle, VM用に予約 */
    NULL,                           /* F_PTR control, port_command のコールバック */
    NULL,                           /* F_PTR timeout, 予約済み */
    NULL,                           /* F_PTR outputv, 予約済み */
    NULL,                           /* F_PTR ready_async, 非同期ドライバ専用 */
    NULL,                           /* F_PTR flush, ポートが閉じられようとしているが
                       ドライバのキューにまだデータが残っているときに呼ばれる */
    NULL,                           /* F_PTR call, control とよく似た、ドライバへの
                       同期呼び出し */
    NULL,                           /* 未使用 */
    ERL_DRV_EXTENDED_MARKER,        /* int extended marker, ドライバのバージョン管理を
                       示すため常にこの値を設定する */
    ERL_DRV_EXTENDED_MAJOR_VERSION, /* int major_version, 常にこの値を
                       設定する */
    ERL_DRV_EXTENDED_MINOR_VERSION, /* int minor_version, 常にこの値を
                       設定する */
    0,                              /* int driver_flags, ドキュメント参照 */
    NULL,                           /* void *handle2, VM用に予約 */
    NULL,                           /* F_PTR process_exit, 監視対象のプロセスが
                       終了したときに呼ばれる */
    NULL                            /* F_PTR stop_select, イベントオブジェクトを
                       閉じるために呼ばれる */
};

DRIVER_INIT(busy_port_drv) /* driver_entry 内の名前と一致させる必要がある */
{
    return &bp_driver_entry;
}
```

これで、この関数をErlangから同期的に呼び出したり、非同期メッセージとして送ったりできるようになりました。ポートハンドラには、ポートへ10個のメッセージを送り、その結果を同期・非同期の両方の方法で受け取るテストも追加しています。

``` erlang
%% https://www.erlang.org/doc/system/c_portdriver.html を元にしている

-module(busy_port).
-export([start/1, stop/0, init/1]).
-export([foo/1, bar/1, async_foo/1, async_bar/1, async_receive/0]).
-export([test_sync_foo/0,
         test_async_foo/0,
         test_async_bar/0]).

start(SharedLib) ->
    case erl_ddll:load_driver(".", SharedLib) of
        ok -> ok;
        {error, already_loaded} -> ok;
        _ -> exit({error, could_not_load_driver})
    end,
    spawn(?MODULE, init, [SharedLib]).

init(SharedLib) ->
    register(busy_port_example, self()),
    Port = open_port({spawn, SharedLib}, []),
    loop(Port, [], 0).

test_sync_foo() ->
    [foo(N) || N <- lists:seq(1, 10)].

test_async_foo() ->
    [async_receive() || _ <- [async_foo(N) || N <- lists:seq(1, 10)]].

test_async_bar() ->
    [async_receive() || _ <- [async_bar(N) || N <- lists:seq(1, 10)]].



stop() ->
    busy_port_example ! stop.

foo(X) ->
    call_port({foo, X}).
bar(Y) ->
    call_port({bar, Y}).

async_foo(X) ->
    send_message({foo, X}).
async_bar(Y) ->
    send_message({bar, Y}).

async_receive() ->
    receive
        {busy_port_example, Data} ->
            Data
    after 2000 -> timeout
    end.

call_port(Msg) ->
    busy_port_example ! {call, self(), Msg},
    receive
    {busy_port_example, Result} ->
        Result
    end.

send_message(Message) ->
    busy_port_example ! {send, self(), Message}.


reply(Id, [{Id, From}|Ids], Data) ->
    From ! {busy_port_example, Data},
    Ids;
reply(Id, [Id1|Ids], Data) ->
    [Id1 | reply(Id, Ids, Data)];
reply(_Id, [], Data) -> %% おっと、対応するIDが見つからない
    io:format("No ID found for data: ~p~n", [Data]),
    [].


loop(Port, Ids, Id) ->
    receive
    {call, Caller, Msg} ->
        io:format("Call: ~p~n", [Msg]),
        Port ! {self(), {command, encode(Msg, Id)}},
        receive
        {Port, {data, Data}} ->
            Res = decode_data(Data),
            io:format("Received data: ~w~n", [Res]),
            Caller ! {busy_port_example, Res}
        end,
        loop(Port, Ids, Id);
     {Port, {data, Data}} ->
            {Ref, Res} = decode(Data),
            io:format("Received data: ~w~n", [Res]),
            NewIds = reply(Ref, Ids, Res),
            loop(Port, NewIds, Id);
     {send, From, Message} ->
            T1 = os:system_time(millisecond),
            io:format("Send: ~p~n", [Message]),
            Port ! {self(), {command, encode(Message, Id)}},
            T2 = os:system_time(millisecond),
            if (T2 - T1) > 500 -> io:format("Shouldn't ! be async...~n", []);
               true -> ok
            end,
            loop(Port, [{Id, From} | Ids], Id + 1);
    stop ->
        Port ! {self(), close},
        receive
        {Port, closed} ->
            exit(normal)
        end;
    {'EXIT', Port, Reason} ->
        io:format("~p ~n", [Reason]),
        exit(port_terminated)
    end.

encode({foo, X}, Id) -> [1, X, Id];
encode({bar, X}, Id) -> [2, X, Id].

decode([Int, Id]) -> {Id, Int}.
decode_data([Int,_Id]) -> Int.
```

シェルで試してみましょう。

``` erlang
1> c(busy_port).
{ok,busy_port}
2> busy_port:start("busy_port_drv").
<0.89.0>
3> busy_port:test_sync_foo().
Call: {foo,1}
Received data: 2
Call: {foo,2}
Received data: 3
Call: {foo,3}
Received data: 4
Call: {foo,4}
Received data: 5
Call: {foo,5}
Received data: 6
Call: {foo,6}
Received data: 7
Call: {foo,7}
Received data: 8
Call: {foo,8}
Received data: 9
Call: {foo,9}
Received data: 10
Call: {foo,10}
Received data: 11
[2,3,4,5,6,7,8,9,10,11]
```

期待どおりに動きました。同期呼び出しを行い、即座に応答を受け取っています。次は非同期呼び出しを試してみましょう。

``` erlang
4> busy_port:test_async_foo().
Send: {foo,1}
Send: {foo,2}
Send: {foo,3}
Send: {foo,4}
Send: {foo,5}
Send: {foo,6}
Send: {foo,7}
Send: {foo,8}
Send: {foo,9}
Send: {foo,10}
Received data: 2
Received data: 3
Received data: 4
Received data: 5
Received data: 6
Received data: 7
Received data: 8
Received data: 9
Received data: 10
Received data: 11
[2,3,4,5,6,7,8,9,10,11]
```

これも期待どおりに動きました。10個のメッセージを送り、結果を送信したのと同じ順序で、こちらも即座に受け取っています。今度はビジーポートを試してみましょう。

``` erlang
5> busy_port:test_async_bar().
Send: {bar,1}
Shouldn't ! be async...
Send: {bar,2}
Shouldn't ! be async...
Send: {bar,3}
Shouldn't ! be async...
Send: {bar,4}
Shouldn't ! be async...
Send: {bar,5}
Shouldn't ! be async...
Send: {bar,6}
Send: {bar,7}
Send: {bar,8}
Send: {bar,9}
Send: {bar,10}
Received data: 2
Received data: 4
Received data: 6
Received data: 8
Received data: 10
Received data: 12
Received data: 14
Received data: 16
Received data: 18
Received data: 20
[timeout,2,4,6,8,10,12,14,16,18]
```

最初の5件は非同期ではなく、残りの5件は非同期になっているのがわかります。これは、ポートがビジーで送信操作がブロックしているためです。ポートがビジーになっているのは、メッセージのIDが14未満で、呼び出し先が`bar`関数だからです。ポートは5秒間ビジーの状態が続き、そのあと残りの5件のメッセージが非同期に送られます。

<a id="_port_scheduling"></a>

### ポートのスケジューリング

Erlangのポートは、プロセスと同様に、TCPのような外部通信を扱うコード（ドライバ）を実行します。もともと、ポートのシグナルは同期的に処理されており、これがI/Oイベントの並列性に問題を引き起こしていました。ロックの競合が大きく、並列性を活かせる余地が減っていたためです。

この問題に対処するため、Erlangはすべてのポートシグナルをスケジューリングし、単一のスケジューラによって順番に実行されるようにしています。これにより競合がなくなり、他のプロセスはErlangコードを並列に実行し続けられます。

ポートは「半ロック」方式で管理されるタスクキューを持っており、公開されたロック付きキューと、非公開のロックフリーキューの2つから成ります。ロックの競合を避けるため、タスクはこれらのキューの間を移動します。この仕組みは、アトミック操作を使ってタスクを中断済みとしてマークすることでI/Oシグナルの中断を扱い、ロックの競合なしに安全にタスクを解放できるようにしています。

ポートは、コマンドシグナルが過負荷になるとビジー状態に入ることがあり、キューが処理できる量に戻るまで新しいシグナルを一時停止します。これにより、ポートが処理しきれる前に押し寄せる過負荷を防ぐフロー制御が実現されています。

シグナルデータの準備はポートのロックを取得する前に行われ、レイテンシを抑えています。競合のないシグナルは即座に実行されて低レイテンシが保たれる一方、競合のあるシグナルは並列性を保つために後回しでスケジューリングされます。

スケジューラの動作と、ポートが全体のスケジューリングの仕組みにどう組み込まれているかの詳細は、[スケジューリングの章](../scheduling/#CH-Scheduling)を参照してください。

<a id="_distributed_erlang"></a>

## 分散Erlang

組み込みの分散レイヤーについての詳細は、[分散Erlangの章](../distribution/#CH-Distribution)を参照してください。

<a id="_sockets_udp_and_tcp"></a>

## ソケット、UDP、TCP

ソケットは、Erlangにおけるネットワーク通信の基本的な要素です。ソケットを使うことで、プロセスはTCPやUDPといったプロトコルでネットワーク越しに通信できます。ここでは、ソケットの扱い方、ソケットに関する情報の取得方法、ソケットの挙動を調整する方法を見ていきます。

Erlangには、ソケットを作成・管理するための堅牢な関数群が用意されています。`gen_tcp`モジュールと`gen_udp`モジュールが、それぞれTCPとUDPプロトコルの利用を助けてくれます。TCPソケットを開く基本的な例を示します。

``` erlang
%% ポート1234でリスニングソケットを開く
{ok, ListenSocket} = gen_tcp:listen(1234, [binary, {packet, 0}, {active, false}, {reuseaddr, true}]),

%% 接続を受け付ける
{ok, Socket} = gen_tcp:accept(ListenSocket),

%% データを送受信する
ok = gen_tcp:send(Socket, <<"Hello, World!">>),
{ok, Data} = gen_tcp:recv(Socket, 0).
```

UDPの場合も手順はよく似ていますが、`gen_udp`モジュールを使います。

``` erlang
%% ポート1234でUDPソケットを開く
{ok, Socket} = gen_udp:open(1234, [binary, {active, false}]),

%% データを送受信する
ok = gen_udp:send(Socket, "localhost", 1234, <<"Hello, World!">>),
receive
    {udp, Socket, Host, Port, Data} -> io:format("Received: ~p~n", [Data])
end.
```

Erlangには、ソケットに関する情報を取得するための関数もいくつか用意されています。たとえば、`inet:getopts/2`と`inet:setopts/2`を使うと、ソケットのオプションを取得・設定できます。例を示します。

``` erlang
%% ソケットのオプションを取得する
{ok, Options} = inet:getopts(Socket, [recbuf, sndbuf, nodelay]),

%% ソケットのオプションを設定する
ok = inet:setopts(Socket, [{recbuf, 4096}, {sndbuf, 4096}, {nodelay, true}]).
```

さらに、`inet:peername/1`と`inet:sockname/1`を使うと、ソケットのリモートアドレスとローカルアドレスを取得できます。

``` erlang
%% 接続済みソケットのリモートアドレスを取得する
{ok, {Address, Port}} = inet:peername(Socket),

%% ソケットのローカルアドレスを取得する
{ok, {LocalAddress, LocalPort}} = inet:sockname(Socket).
```

ソケットの性能や挙動を最適化するために、さまざまなソケットオプションを調整できます。よく調整されるオプションには、バッファサイズ、タイムアウト、パケットサイズなどがあります。これらのオプションを調整する例を示します。

``` erlang
%% ソケットのバッファサイズを設定する
ok = inet:setopts(Socket, [{recbuf, 8192}, {sndbuf, 8192}]),

%% データ受信のタイムアウトを設定する
ok = inet:setopts(Socket, [{recv_timeout, 5000}]),

%% TCPソケットのパケットサイズを設定する
ok = inet:setopts(Socket, [{packet, 4}]).
```