Erlang Tcp 接受者模式

Question

考虑以下（基于 LYSE 的 sockserv）

%%% The supervisor in charge of all the socket acceptors.
-module(tcpsocket_sup).
-behaviour(supervisor).

-export([start_link/0, start_socket/0]).
-export([init/1]).

start_link() ->
  supervisor:start_link({local, ?MODULE}, ?MODULE, []).

init([]) ->
  {ok, Port} = application:get_env(my_app,tcpPort),
  {ok, ListenSocket} = gen_tcp:listen(
    Port,
    [binary, {packet, 0}, {reuseaddr, true}, {active, true} ]),
  lager:info(io_lib:format("Listening for TCP on port ~p", [Port])),
  spawn_link(fun empty_listeners/0),
  {ok, {{simple_one_for_one, 60, 3600},
    [{socket,
      {tcpserver, start_link, [ListenSocket]},
      temporary, 1000, worker, [tcpserver]}
    ]}}.

start_socket() ->
  supervisor:start_child(?MODULE, []).%,

empty_listeners() ->
  [start_socket() || _ <- lists:seq(1,20)],
  ok.

---

%%%-------------------------------------------------------------------
%%% @author mylesmcdonnell
%%% @copyright (C) 2015, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 06. Feb 2015 07:49
%%%-------------------------------------------------------------------
-module(tcpserver).
-author("mylesmcdonnell").

-behaviour(gen_server).

-record(state, {
    next,
    socket}).

-export([start_link/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, code_change/3, terminate/2]).

-define(SOCK(Msg), {tcp, _Port, Msg}).
-define(TIME, 800).
-define(EXP, 50).

start_link(Socket) ->
  gen_server:start_link(?MODULE, Socket, []).

init(Socket) ->
  gen_server:cast(self(), accept),
  {ok, #state{socket=Socket}}.

handle_call(_E, _From, State) ->
  {noreply, State}.

handle_cast(accept, S = #state{socket=ListenSocket}) ->
  {ok, AcceptSocket} = gen_tcp:accept(ListenSocket),
  kvstore_tcpsocket_sup:start_socket(),
  receive
    {tcp, Socket, <<"store",Value/binary>>} ->
      Uid = kvstore:store(Value),
      send(Socket,Uid);
    {tcp, Socket, <<"retrieve",Key/binary>>} ->
      case kvstore:retrieve(binary_to_list(Key)) of
        [{_, Value}|_] ->
          send(Socket,Value);
        _ ->
          send(Socket,<<>>)
      end;
    {tcp, Socket, _} ->
      send(Socket, "INVALID_MSG")
  end,
  {noreply, S#state{socket=AcceptSocket, next=name}}.

handle_info(_, S) ->
  {noreply, S}.

code_change(_OldVsn, State, _Extra) ->
  {ok, State}.

terminate(normal, _State) ->
  ok;
terminate(_Reason, _State) ->
  lager:info("terminate reason: ~p~n", [_Reason]).

send(Socket, Bin) ->
  ok = gen_tcp:send(Socket, Bin),
  ok = gen_tcp:close(Socket),
  ok.

我不清楚每个 tcpserver 进程是如何终止的？这是泄漏进程吗？

Answer 1

我没有看到您正在终止拥有进程的任何地方。

我想你要找的是四种情况：

• 客户端终止连接（您收到tcp_closed）
• 连接不稳定（您会收到tcp_error）
• 服务器收到要终止的系统消息（当然，这可能只是主管将其杀死，或者是终止消息）
• 客户端向服务器发送一条消息，告诉服务器它已完成，除了对tcp_closed 做出反应之外，您还想做一些清理工作。

最常见的情况通常是客户端只是关闭连接，为此您需要类似：

handle_info({tcp_closed, _}, State) ->
    {stop, normal, State};

连接变得奇怪总是有可能的。我想不出任何时候我想要拥有进程或套接字，所以：

%% You might want to log something here.
handle_info({tcp_error, _}, State) ->
    {stop, normal, State};

任何情况下，客户端告诉服务器它已经完成并且您需要根据客户端已经成功完成某些操作来进行清理（可能您打开了应该首先写入的资源，或者打开了待处理的数据库事务，或者其他任何情况） ) 您会希望客户端发出一条成功消息，该消息会像您的 send/2 那样关闭连接，并返回 {stop, normal, State} 以停止该过程。

这里的关键是确保您确定要终止连接并杀死服务器进程或（更好）返回{stop, Reason, State}的情况。

如上所述，如果您希望 send/2 成为单个响应和干净的退出（或者实际上，每个 accept 转换应该导致单个 send/2 然后终止），那么您想要：

handle_cast(accept, S = #state{socket=ListenSocket}) ->
  {ok, AcceptSocket} = gen_tcp:accept(ListenSocket),
  kvstore_tcpsocket_sup:start_socket(),
  receive
    %% stuff that results in a call to send/2 in any case.
  end,
  {stop, normal, S}.

LYSE 演示的情况是连接是持久的，并且客户端和服务器之间存在持续的来回来回。在上述情况下，您正在处理单个请求，生成一个新的侦听器以重新填充侦听器池，并且应该退出，因为您没有计划让这个 gen_server 做任何进一步的工作。