使用 OTP 原则的非阻塞 TCP 服务器

Posted

技术标签:

【中文标题】使用 OTP 原则的非阻塞 TCP 服务器【英文标题】:Non-blocking TCP server using OTP principles 【发布时间】:2011-09-04 11:08:54 【问题描述】:

我开始学习 Erlang,所以我想写“你好,世界!”并发编程,一个 IRC 机器人。

我已经使用 Erlang 编写了一个,没有任何 OTP 细节(主管、应用程序等行为)。我希望使用 OTP 原则重写它,但不幸的是我无法找出使用 OTP 进行套接字编程的“正确”方法。

似乎唯一合理的方法是手动创建另一个进程并将其链接到主管,但肯定有人在某个地方曾经这样做过。

【问题讨论】:

【参考方案1】:

实现异步 TCP 侦听器的另一种方法是使用 supervisor_bridge

这是我编写的一些代码来显示这一点(未经测试):

-module(connection_bridge).

-behaviour(supervisor_bridge).

% supervisor_bridge export
-export([init/1, terminate/2]).

% internal proc_lib:start_link
-export([accept_init/3]).

%% Port: see gen_tcp:listen(Port, _).
%% Options: see gen_tcp:listen(_, Options).
%% ConnectionHandler: Module:Function(Arguments)->pid() or fun/0->pid()
%% ConnectionHandler: return pid that will receive TCP messages
init(Port, Options, ConnectionHandler) ->
    case gen_tcp:listen(Port, Options) of
        ok, ListenSocket ->
            ok, ServerPid = proc_lib:start_link(?MODULE, accept_init,
                [self(), ListenSocket, ConnectionHandler], 1000),
            ok, ServerPid, ListenSocket;
        OtherResult -> OtherResult
    end.

terminate(_Reason, ListenSocket) ->
    gen_tcp:close(ListenSocket).

accept_init(ParentPid, ListenSocket, ConnectionHandler) ->
    proc_lib:init_ack(ParentPid, ok, self()),
    accept_loop(ListenSocket, ConnectionHandler).

accept_loop(ListenSocket, ConnectionHandler) ->
    case gen_tcp:accept(ListenSocket) of
        ok, ClientSocket ->
            Pid = case ConnectionHandler of
                Module, Function, Arguments ->
                    apply(Module, Function, Arguments);
                Function when is_function(Function, 0) ->
                    Function()
            end,
            ok = gen_tcp:controlling_process(ClientSocket, Pid),
            accept_loop(ListenSocket, ConnectionHandler);
        error, closed ->
            error(shutdown, tcp_closed);
        error, Reason ->
            error(Reason)
    end.

比我的其他答案更容易理解。 connection_bridge 也可以扩展为支持 UDP 和 SCTP。

【讨论】:

【参考方案2】:

很高兴您开始学习 Erlang/OTP!

以下资源非常有用:

OTP Design Principles。如果您还没有,请仔细阅读。请注意 OTP 是面向对象 (OO) 的常见误解:它不是!忘记关于“继承”的一切。仅仅通过“扩展”标准模块来构建完整的系统是不可能的。 Messaging System:

这些函数必须用于实现一个进程使用系统消息

Special Processes。特殊流程是符合 OTP 的流程,可以与主管很好地集成。

这是我项目中的一些代码。我也是 Erlang 学习者,所以请不要太相信代码。

-module(gen_tcpserver).

%% Public API
-export([start_link/2]).

%% start_link reference
-export([init/2]).

%% System internal API
-export([system_continue/3, system_terminate/4, system_code_change/4]).

-define(ACCEPT_TIMEOUT, 250).

-record(server_state, socket=undefined,
                       args,
                       func).

%% ListenArgs are given to gen_tcp:listen
%% AcceptFun(Socket) -> ok, blocks the TCP accept loop
start_link(ListenArgs, AcceptFun) ->
    State = #server_stateargs=ListenArgs,func=AcceptFun,
    proc_lib:start_link(?MODULE, init, [self(), State]).

init(Parent, State) ->
    Port, Options = State#server_state.args,
    ok, ListenSocket = gen_tcp:listen(Port, Options),
    NewState = State#server_statesocket=ListenSocket,
    Debug = sys:debug_options([]),
    proc_lib:init_ack(Parent, ok, self()),
    loop(Parent, Debug, NewState).

loop(Parent, Debug, State) ->
    case gen_tcp:accept(State#server_state.socket, ?ACCEPT_TIMEOUT) of
        ok, Socket when Debug =:= [] -> ok = (State#server_state.func)(Socket);
        ok, Socket ->
            sys:handle_debug(Debug, fun print_event/3, undefined, accepted, Socket),
            ok = (State#server_state.func)(Socket);
        error, timeout -> ok;
        error, closed when Debug =:= [] ->
            sys:handle_debug(Debug, fun print_event/3, undefined, closed),
            exit(normal);
        error, closed -> exit(normal)
    end,
    flush(Parent, Debug, State).

flush(Parent, Debug, State) ->
    receive
        system, From, Msg ->
            sys:handle_system_msg(Msg, From, Parent, ?MODULE, Debug, State)
        after 0 ->
            loop(Parent, Debug, State)
    end.

print_event(Device, Event, _Extra) ->
    io:format(Device, "*DBG* TCP event = ~p~n", [Event]).

system_continue(Parent, Debug, State) ->
    loop(Parent, Debug, State).

system_terminate(Reason, _Parent, _Debug, State) ->
    gen_tcp:close(State#server_state.socket),
    exit(Reason).

system_code_change(State, _Module, _OldVsn, _Extra) ->
    ok, State.

请注意,这是一个合规的 OTP 流程(可由主管管理)。你应该使用AcceptFun 来产生(=更快)一个新的工人孩子。不过我还没有彻底测试过。

1> ok, A = gen_tcpserver:start_link(8080,[],fun(Socket)->gen_tcp:close(Socket) end).
ok,<0.93.0>
2> sys:trace(A, true).
ok
*DBG* TCP event = accepted,#Port<0.2102>
*DBG* TCP event = accepted,#Port<0.2103>
3> 

(在2&gt;ok 之后,我将Google Chrome 浏览器指向了8080 端口:对TCP 的一次很好的测试!)

【讨论】:

顺便说一下,使用这个 OTP 过程,任何主管的 shutdown 参数都是有意义的:它至少应该大于 ACCEPT_TIMEOUT【参考方案3】:

我认为这就是您要寻找的: http://www.trapexit.org/Building_a_Non-blocking_TCP_server_using_OTP_principles 这是一个关于如何使用 OTP 构建非阻塞 TCP 服务器的完整教程(当然,有完整的文档和解释)。

【讨论】:

不,使用未记录(且可能不稳定)的 prim_inet:async_accept/2。也许没有“OTP 方式”可以做到这一点:/ 在这种情况下,我将简单地使用 gen_tcp:accept/1 和 gen_tcp:controlling_process/2 (如文档所示:“将新的控制进程 Pid 分配给 Socket。控制进程是从套接字接收消息。如果被当前控制进程以外的任何其他进程调用,则返回 error, eperm。")。这是一个如何使用它的示例:20bits.com/articles/erlang-a-generalized-tcp-server(请注意以下段落:“使用 gen_server 实现网络服务器的问题是调用 gen_tcp:accept ...”。希望这会有所帮助。

以上是关于使用 OTP 原则的非阻塞 TCP 服务器的主要内容,如果未能解决你的问题,请参考以下文章

NIO实现TCP的非阻塞通信

关闭tcp套接字的erlang

Posix PIPE 上的非阻塞读取和阻塞写入

DLL 中的非阻塞套接字(无窗口)

为啥在 Erlang 中使用 OTP?

未发出信号的非阻塞 ConnectNamedPipe 事件