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

-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}.

-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).

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>

1> group_leader().
<0.24.0>

2> self().
<0.42.0>

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

erl -sname mynode -setcookie mycookie -detached

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

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

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

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

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

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

/* example.c */

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

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

/* 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;
}

%% 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.

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]

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]

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]

%% ポート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).

%% ポート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.

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

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

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

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

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

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

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