今天有同學在gmail裡面問了一個Erlang的問題，問題描述的非常好, 如下：

問題的背景是：

1、我開發了一個服務端程式，接收客戶端的連線。同一時刻會有多個客戶端來連線，連線後，接收客戶端請求後，再發送響應訊息，然後客戶端主動斷連。

2、服務端監聽的socket屬性設定如下：

[binary, {packet, raw},
{ip, IPAddr}, {backlog, 10000},
{active, false}, {reuseaddr, true},
{nodelay, false}, {delay_send, true},
{recbuf, 128 * 1024}, {sndbuf, 64 * 1024}]

3、伺服器accept監聽socket，接收客戶端請求，傳送響應訊息分別是在3個不同的程序中進行。接收請求和傳送響應的程序都是重複使用的，每次重新使用的時候傳入一個新accept的socket。

問題的現象是：

1、單個使用者發起呼叫的時候，流程是成功的，伺服器能正常響應。但是多個使用者一起呼，批量跑的時候，跑一段時間後，部分客戶端會發現不能接收到伺服器返回的響應。從抓包來看，客戶端的請求是傳送到伺服器端了。

2、伺服器這邊傳送響應的程序會收到一條{empty_out_q, #Port<0.25876>}這樣的訊息，而這條訊息並不是我開發的程式碼產生的。

問題是：

1、為什麼發訊息的程序會收到{empty_out_q, #Port<0.25876>}這樣的訊息？

2、收到empty_out_q訊息，是不是就說明呼叫gen_tcp:send傳送失敗？

3、是不是說設定了delay_send的屬性，所以即使send失敗，也是非同步的，在呼叫send的時候會馬上返回ok，但是後面真的傳送失敗後，則系統會給呼叫send方法的程序傳送一條{empty_out_q, #Port<0.25876>}這樣的訊息。

這個問題非常有意思。

首先我來解釋下erts內部inet_drv工作原理：gen_tcp:send的時候在正常情況下，底層的驅動會馬上呼叫作業系統的send去傳送資料，如果一次沒發完資料，會把資料暫存在驅動的傳送佇列裡面，同時向epoll等事件檢查器登記socket寫事件，然後等待socket的可寫事件的發生。 delay_send選項的的作用是不嘗試立即傳送，而是把資料都暫存在驅動的傳送佇列，然後等待可寫, 從而可以批量傳送資料，提高效率。

有了這樣的機制後，那麼在gen_tcp:close或者shutdown的時候，要保證這些暫存的資料先發送完成。那麼在知道目前還有暫存的資料未傳送的時候，不會立即執行close或者shutdown操作，而是會先等待這些資料的傳送。

等待過程是這樣的，先把呼叫者程序的pid提交到socket的empty_out_q等待佇列去，同時返回目前暫存的資料的位元組數。這時候底層的傳送過程還在繼續，如果在傳送的時候，發現數據都發生完畢，就給呼叫者程序傳送一個{empty_out_q, Port}訊息，表明可以進行下一步動作。

參看程式碼： otp_src_R14B03/erts/preloaded/src/prim_inet.erl: L117

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%
%% Shutdown(insock(), atom()) -> ok
%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% TODO: shutdown equivalent for SCTP
%%
shutdown(S, read) when is_port(S) ->
    shutdown_2(S, 0);
shutdown(S, write) when is_port(S) ->
    shutdown_1(S, 1);
shutdown(S, read_write) when is_port(S) ->
    shutdown_1(S, 2).

shutdown_1(S, How) ->
    case subscribe(S, [subs_empty_out_q]) of
        {ok,[{subs_empty_out_q,N}]} when N > 0 ->
            shutdown_pend_loop(S, N);   %% wait for pending output to be sent
        _Other -> ok
    end,
    shutdown_2(S, How).

shutdown_2(S, How) ->
    case ctl_cmd(S, ?TCP_REQ_SHUTDOWN, [How]) of
   {ok, []} -> ok;
        Error -> Error
    end.

shutdown_pend_loop(S, N0) ->
    receive
        {empty_out_q,S} -> ok
    after ?INET_CLOSE_TIMEOUT ->
            case getstat(S, [send_pend]) of
                {ok,[{send_pend,N0}]} -> ok;
                {ok,[{send_pend,N}]} -> shutdown_pend_loop(S, N);
                _ -> ok
            end
    end.

如果超過5秒，都沒有收到{empty_out_q, Port}訊息,那麼就看下目前的暫存的資料的位元組數, 並且和最初的暫存的資料的位元組數比較，如果一直沒變的話，那麼說明由於種種原因，把這些資料傳送出去比較沒希望，那麼他就果斷的繼續下一步動作。如果位元組數在變少的話，那麼就繼續等。

現在問題來了。shutdown 在發現暫存的資料沒有希望發出去的時候，選擇不作為，那麼作為後遺症，caller被登記在通知名單裡面。過一段時間，要不資料真的被髮出去了，要不就是發生以下事件： 1. socket發現讀錯誤。2. socket發現寫錯誤。 3. 對端關閉。 4. 宿主程序退出，socket作為一個port被強行關閉，這時候需要清空傳送緩衝區，這時候會同時給caller傳送{empty_out_q, Port}訊息。

有圖有真相，我們讓程式碼說話。

參看程式碼： otp_src_R14B03/erts/emulator/drivers/common/inet_drv.c:

// L7100 這個函式負責給呼叫者傳送訊息empty_out_q
static void
send_empty_out_q_msgs(inet_descriptor* desc)
{
  ErlDrvTermData msg[6];
  int msg_len = 0;

  if(NO_SUBSCRIBERS(&desc->empty_out_q_subs))
    return;

  msg_len = LOAD_ATOM(msg, msg_len, am_empty_out_q);
  msg_len = LOAD_PORT(msg, msg_len, desc->dport);
  msg_len = LOAD_TUPLE(msg, msg_len, 2);

  ASSERT(msg_len == sizeof(msg)/sizeof(*msg));

  send_to_subscribers(desc->port,
                      &desc->empty_out_q_subs,
                      1,
                      msg,
                      msg_len);
}

//那麼誰會呼叫他呢: tcp_clear_output和tcp_inet_output.

static int tcp_inet_output(tcp_descriptor* desc, HANDLE event)
{
...
else if (IS_CONNECTED(INETP(desc))) {
        for (;;) {
            int vsize;
            int n;
            SysIOVec* iov;

            if ((iov = driver_peekq(ix, &vsize)) == NULL) {
                sock_select(INETP(desc), FD_WRITE, 0);
                send_empty_out_q_msgs(INETP(desc));
                goto done;
            }
            vsize = vsize > MAX_VSIZE ? MAX_VSIZE : vsize;
            DEBUGF(("tcp_inet_output(%ld): s=%d, About to send %d items\r\n",
                    (long)desc->inet.port, desc->inet.s, vsize));
            if (IS_SOCKET_ERROR(sock_sendv(desc->inet.s, iov, vsize, &n, 0))) {
                if ((sock_errno() != ERRNO_BLOCK) && (sock_errno() != EINTR)) {
                    DEBUGF(("tcp_inet_output(%ld): sock_sendv(%d) errno = %d\r\n",
                            (long)desc->inet.port, vsize, sock_errno()));
                    ret =  tcp_send_error(desc, sock_errno());
                    goto done;
                }
#ifdef __WIN32__
                desc->inet.send_would_block = 1;
#endif
                goto done;
...
}

//以及：

/* clear QUEUED output */
static void tcp_clear_output(tcp_descriptor* desc)
{
    ErlDrvPort ix  = desc->inet.port;
    int qsz = driver_sizeq(ix);

    driver_deq(ix, qsz);
    send_empty_out_q_msgs(INETP(desc));
}

//誰呼叫 tcp_clear_output呢？
//tcp_inet_flush,
//tcp_recv_closed,
//tcp_recv_error,
//tcp_send_error,

//特別是tcp_recv_closed函式

/* The peer socket has closed, cleanup and send event */
static int tcp_recv_closed(tcp_descriptor* desc)
{
#ifdef DEBUG
    long port = (long) desc->inet.port; /* Used after driver_exit() */
#endif
    DEBUGF(("tcp_recv_closed(%ld): s=%d, in %s, line %d\r\n",
            port, desc->inet.s, __FILE__, __LINE__));
    if (IS_BUSY(INETP(desc))) {
        /* A send is blocked */
        desc->inet.caller = desc->inet.busy_caller;
        tcp_clear_output(desc);
        if (desc->busy_on_send) {
            driver_cancel_timer(desc->inet.port);
            desc->busy_on_send = 0;
            DEBUGF(("tcp_recv_closed(%ld): busy on send\r\n", port));
        }
        desc->inet.state &= ~INET_F_BUSY;
        set_busy_port(desc->inet.port, 0);
        inet_reply_error_am(INETP(desc), am_closed);
        DEBUGF(("tcp_recv_closed(%ld): busy reply 'closed'\r\n", port));
    }
    if (!desc->inet.active) {
        /* We must cancel any timer here ! */
        driver_cancel_timer(desc->inet.port);
        /* passive mode do not terminate port ! */
        tcp_clear_input(desc);
        if (desc->inet.exitf) {
            tcp_clear_output(desc);
            desc_close(INETP(desc));
        } else {
            desc_close_read(INETP(desc));
        }
        async_error_am_all(INETP(desc), am_closed);
        /* next time EXBADSEQ will be delivered  */
        DEBUGF(("tcp_recv_closed(%ld): passive reply all 'closed'\r\n", port));
    } else {
        tcp_clear_input(desc);
        tcp_closed_message(desc);
        if (desc->inet.exitf) {
            driver_exit(desc->inet.port, 0);
        } else {
            desc_close_read(INETP(desc));
        }
        DEBUGF(("tcp_recv_closed(%ld): active close\r\n", port));
    }
    DEBUGF(("tcp_recv_closed(%ld): done\r\n", port));
    return -1;
}
/* 我們看到在：
1. 對端關閉的時候，
2. 而且我端是被動接收，
3. socket開啟這二個選項的時候最容易{exit_on_close, true}, {delay_send,true} 
最容易發生上面的現象
*/

只有這種情況你的程序會收到該訊息。

從你的描述來看

“同一時刻會有多個客戶端來連線，連線後，接收客戶端請求後，再發送響應訊息，然後客戶端主動斷連。”.

基本上可以推斷是，你用shutdown(write 或者 read_write)後，你的客戶端由於某種原因斷鏈，你的傳送程序收到這樣的訊息。

我們來試驗下我們的猜想：

$ cat test.erl
-module(test).
-compile(export_all).

start() ->
    start(1234).

start(Port) ->
    register(?MODULE, self()),
    
    spawn_link(fun ()-> S= listen(Port), accept(S) end),

    receive Any -> io:format("~p~n", [Any]) end.  %% to stop: test!stop.

listen(Port) ->
    Opts = [{active, false},
            binary,
            {backlog, 256},
            {packet, raw},
            {reuseaddr, true}],

    {ok, S} = gen_tcp:listen(Port, Opts),
    S.

accept(S) ->
    case gen_tcp:accept(S) of
        {ok, Socket} -> inet:setopts(Socket, [{exit_on_close, true},    
                                              {delay_send,true}]), 
                        spawn_opt(?MODULE, entry, [Socket], []);
        Error    -> erlang:error(Error)
    end,
    accept(S).


entry(S)->
        loop(S),
        check_empty_out_q_msg(1000),
        io:format("bye socket ~p~n",[S]),
        ok.


check_empty_out_q_msg(Timeout)->
    receive 
        Any -> io:format("bingo, got message ~p~n", [Any]), Any
    after Timeout -> cont end.



loop(S) ->
    check_empty_out_q_msg(0),
    
    case gen_tcp:recv(S, 0) of
        {ok, <<"start", _/binary>>}->
                io:format("start to reproduce {empty_out_q, Port} message ~n",[]),
                gen_tcp:send(S, lists:duplicate(1024*1024, "A")),
                io:format("sent 1M bytes ~n",[]),
                
                io:format("sleep 1s ~n",[]),
                receive Any1 -> Any1 after 1000 -> cont end,

                loop(S);

        {ok, _Data} ->
            io:format("shutdown(write) ~n",[]),
            {ok, [{send_pend, N}]}=inet:getstat(S, [send_pend]),
            gen_tcp:shutdown(S, write),
            {ok, [{send_pend, N1}]}=inet:getstat(S, [send_pend]),
            io:format("5s send_pend ~w/~w ~n",[N,N1]),
            loop(S);
            
        Error ->
            io:format("tcp ~p~n", [Error]),
            Error
    end.

$ cat client.erl
-module(client).
-export([start/0]).

start()->
        {ok,Sock} = gen_tcp:connect("localhost", 1234, [{active,false}]),

        gen_tcp:send(Sock, "start"),                                     
        io:format("send start~n",[]),

        gen_tcp:recv(Sock,1024),
        io:format("drain 1024 bytes~n",[]),                                   

        gen_tcp:send(Sock, "bang"),  
        io:format("send bang~n",[]),                                   

        io:format("sleep 10s~n",[]),                                   
        receive
                Any -> Any
        after 10000 -> cont end,

        gen_tcp:shutdown(Sock, write),

        io:format("end~n",[]),                                   
        ok.

$ erlc test.erl client.erl

在一個終端執行：

$ erl -noshell -s test
start to reproduce {empty_out_q, Port} message 
sent 1M bytes 
sleep 1s 
shutdown(write) 
5s send_pend 851968/851968 
tcp {error,closed}
bingo, got message {empty_out_q,#Port<0.456>}
bye socket #Port<0.456>

在另外一個終端執行：

$ erl -noshell -s client
send start
drain 1024 bytes
send bang
sleep 10s
end

中間我們可以看到：

$ ss 
State       Recv-Q Send-Q                                                 Local Address:Port                                                     Peer Address:Port   
ESTAB       0      130944                                                     127.0.0.1:search-agent                                                     127.0.0.1:43273   
...

傳送佇列大量的資料阻塞。

結果驗證了我們的猜想！這個故事告訴我們：原始碼是最權威的。

祝大家玩得開心！

Post Footer automatically generated by wp-posturl plugin for wordpress.