关于erlang socket被动模式和delay_send合用的问题

转载请注明，来自：http://blog.csdn.net/skyman_2001

有项目反应服务器遇到ports()里的port，其port_info/1返回是undefined的问题，而且出现这个后，erlang:halt()不能正常关闭节点，要用erlang:halt(Status, [{flush, false}])才能关闭节点。在很多客户端同时关闭时比较容易重现。我在网上erlang论坛上也发现有人遇到这个问题求助，所以决定研究一下。

通过对比分析ERTS的ports_0()（对应erlang:ports/0）和port_info_1()（对应erlang:port_info/1）的源代码发现，如果一个port的状态是ERTS_PORT_SFLG_CLOSING，则该port会包含在erlang:ports/0返回的列表中，但erlang:port_info/1返回undefined。ERTS_PORT_SFLG_CLOSING是port的中间状态，wqtn22在http://wqtn22.iteye.com/blog/1765741一文中对此有详细的分析。

ERTS源代码（erts/emulator/beam/io.c）中关于它的注释说明如下：

**ERTS_PORT_SFLG_CLOSING is a state where the port is in Limbo, waiting to

** passon. All links are removed, and the port receives in/out-put events so

** assoon as the port queue gets empty terminate_port() is called.

先讲下delay_send，如果打开了delay_send，每个port会维护一个发送队列，数据不是立即发送，而是存到发送队列里，等socket可写的时候再发送，相当于是ERTS自己实现的组包机制。关于Erlang Tcp Send的深度解刨，请看霸爷的这篇文章：http://blog.yufeng.info/archives/336

那么为什么port会停留在ERTS_PORT_SFLG_CLOSING状态呢？通过分析和调试ERTS代码，得出了产生这个问题的过程：

1. 当Socket客户端关闭时，会调用到tcp_recv_closed()函数（见erts/emulator/drivers/common/inet_drv.c），里面：

 

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): passivereply all 'closed'\r\n", port));

        printf("tcp_recv_closed(%ld): passivereply all 'closed', state=%d\r\n", (long) desc->inet.port,desc->inet.state);

    }

 

我们用的是{active,false}，即被动模式。会调用tcp_clear_output()清空发送队列里的数据，最后会向调用进程发送'closed'消息（async_error_am_all(INETP(desc), am_closed)），但是被动模式下是不会terminate port的，所以这时该socket的状态还是INET_STATE_CONNECTED（在宿主进程退出时terminate port，才将该socket状态设为INET_STATE_CLOSED）。

2.  这之后如果向这个port发送数据，看相关代码（inet_drv.c）：

static void tcp_inet_commandv(ErlDrvData e,ErlIOVec* ev)

{

    tcp_descriptor* desc =(tcp_descriptor*)e;

    desc->inet.caller =driver_caller(desc->inet.port);

 

   DEBUGF(("tcp_inet_commanv(%ld) {s=%d\r\n",

           (long)desc->inet.port, desc->inet.s));

    if (!IS_CONNECTED(INETP(desc))) {

        if(desc->tcp_add_flags & TCP_ADDF_DELAYED_CLOSE_SEND) {

           desc->tcp_add_flags &= ~TCP_ADDF_DELAYED_CLOSE_SEND;

           printf("tcp_inet_commandv(%ld) closed\r\n",(long)desc->inet.port);

           inet_reply_error_am(INETP(desc), am_closed);

        }

        else

           inet_reply_error(INETP(desc), ENOTCONN);

    }

    else if (tcp_sendv(desc, ev) == 0)

       inet_reply_ok(INETP(desc));

   DEBUGF(("tcp_inet_commandv(%ld) }\r\n", (long)desc->inet.port));

}

 

上面说了，此时socket的状态是INET_STATE_CONNECTED，所以会执行tcp_sendv()，再跟到tcp_sendv()里面去看：

static int tcp_sendv(tcp_descriptor* desc, ErlIOVec* ev)

{

   ...

 

    if ((sz = driver_sizeq(ix)) > 0) {

        driver_enqv(ix, ev, 0);

        if (sz+ev->size >= desc->high) {

            DEBUGF(("tcp_sendv(%ld):s=%d, sender forced busy\r\n",

                   (long)desc->inet.port, desc->inet.s));

 

            desc->inet.state |=INET_F_BUSY;  /* mark for low-watermark */

            desc->inet.busy_caller = desc->inet.caller;

           set_busy_port(desc->inet.port, 1);

            if (desc->send_timeout !=INET_INFINITY) {

               desc->busy_on_send = 1;

               driver_set_timer(desc->inet.port, desc->send_timeout);

            }

            return 1;

        }

    }

    else {

        int vsize = (ev->vsize > MAX_VSIZE) ?MAX_VSIZE : ev->vsize;

 

        DEBUGF(("tcp_sendv(%ld): s=%d, about tosend "LLU","LLU" bytes\r\n",

               (long)desc->inet.port, desc->inet.s, (llu_t)h_len, (llu_t)len));

        

        if (INETP(desc)->is_ignored) {

            INETP(desc)->is_ignored |=INET_IGNORE_WRITE;

            n = 0;

        } else if(desc->tcp_add_flags & TCP_ADDF_DELAY_SEND) {

            n = 0;

        } else if (IS_SOCKET_ERROR(sock_sendv(desc->inet.s,ev->iov,

                                            vsize, &n, 0))) {

            if ((sock_errno() !=ERRNO_BLOCK) && (sock_errno() != EINTR)) {

                int err =sock_errno();

               DEBUGF(("tcp_sendv(%ld): s=%d, "

                       "sock_sendv(size=2) errno = %d\r\n",

                       (long)desc->inet.port, desc->inet.s, err));

 

                return tcp_send_error(desc, err);

            }

#ifdef __WIN32__

            desc->inet.send_would_block =1;

#endif

            n = 0;

        }

        else if (n == ev->size) {

           ASSERT(NO_SUBSCRIBERS(&INETP(desc)->empty_out_q_subs));

            return 0;

        }

        else {

            DEBUGF(("tcp_sendv(%ld):s=%d, only sent "

                   LLU"/%d of "LLU"/%d bytes/items\r\n",

                   (long)desc->inet.port, desc->inet.s,

                   (llu_t)n, vsize, (llu_t)ev->size, ev->vsize));

        }

 

        DEBUGF(("tcp_sendv(%ld): s=%d, Sendfailed, queuing\r\n",

               (long)desc->inet.port, desc->inet.s));

       

        driver_enqv(ix, ev, n);

        if (!INETP(desc)->is_ignored)

           sock_select(INETP(desc),(FD_WRITE|FD_CLOSE), 1);

    }

    return 0;

}

 

因为sz = driver_sizeq(ix) == 0，所以进入到下面的分支；因为设了delay_send，所以跳到红色的那行分支，也就是不会调用sock_sendv()。到最后执行 driver_enqv(ix, ev, n)，就向发送队列添加数据了。

 

3. 在该socket的宿主进程退出时，会依次调用terminate_proc()-> erts_do_exit_process() -> continue_exit_process()-> erts_sweep_links() -> doit_exit_link()-> erts_do_exit_port()，我们看下erts_do_exit_port()函数里面相关代码：

...

 if((reason != am_kill) && !is_port_ioq_empty(p)) {

      erts_port_status_bandor_set(p,

                                 ~ERTS_PORT_SFLG_EXITING, /* must turn it off*/

                                 ERTS_PORT_SFLG_CLOSING);

     flush_port(p);

   }

  else {

      terminate_port(p);

   }

上面因为发送队列里有数据，不为空，所以将port的状态设为 ERTS_PORT_SFLG_CLOSING，问题就这么产生了！

总结：在被动模式下，若Socket对端关闭，本端不会terminate port，如果在这之后还向该Socket发送数据，则因为设置的是delay_send，所以不会调用sock_sendv()真正发送数据，而是存到发送队列里。最后在该Socket宿主进程退出时，会terminate port，但发现发送队列不为空，所以就将该port设为ERTS_PORT_SFLG_CLOSING。

这个问题只有在被动模式和delay_send合用的时候才会出现，因为如果是非被动模式（active true或once），则对端关闭时，本端会terminate port，后面向该Socket发送数据会出现closed或ENOTCONN错误；同样如果没设置delay_send，则向该socket发送数据时会调用sock_sendv()直接发送数据，因为对端已经关闭，会触发tcp_send_error()，里面会调用erl_inet_close()关闭，这样该Socket状态会变为INET_STATE_CLOSED。

 

因此，要想避免这个问题，不要在对端关闭后再向该Socket发送数据；或者关掉delay_send；或者用非被动模式，比如once。

最后的问题，节点内如果有port状态为ERTS_PORT_SFLG_CLOSING，为什么erlang:halt/0不能关闭节点呢？我们跟踪一下halt的流程：

Erlang:halt/0对应的bif是halt_0()，代码见erts/emulator/beam/bif.c：

/* stop the system */

/* ARGSUSED */

BIF_RETTYPE halt_0(BIF_ALIST_0)

{

   VERBOSE(DEBUG_SYSTEM,("System halted by BIF halt()\n"));

   erl_halt(0);

   ERTS_BIF_YIELD1(bif_export[BIF_halt_1], BIF_P, am_undefined);

}

里面会调用erl_halt()，然后会关闭节点的ports，处理函数是handle_reap_ports()，里面有个引用计数：erts_halt_progress，初始值设为1，然后每个port在关闭前加1，关闭（erts_do_exit_port）后减1。如果每个port都这么有加有减的话，关闭完所有port后erts_halt_progress值还为1，接着将erts_halt_progress减1，若结果为0的话，就退出节点：

if (erts_smp_atomic32_dec_read_nob(&erts_halt_progress) == 0) {

           erl_exit_flush_async(erts_halt_code, "");

        }

erl_exit_flush_async()函数就是执行节点最终退出流程的。

但是如果有port状态是ERTS_PORT_SFLG_CLOSING，那么就不会调用erts_do_exit_port()：

          ...

          if (prt->status &(ERTS_PORT_SFLG_EXITING

                              | ERTS_PORT_SFLG_CLOSING)) {

               erts_port_release(prt);

                continue; //这里continue了

            }

   

            erts_do_exit_port(prt,prt->id, am_killed);

            erts_port_release(prt);

           ...

 

这样引用计数erts_halt_progress就加了没减，这样所有port处理完后erts_halt_progress就不为1，也就不会调用 erl_exit_flush_async()退出了。

那么为什么erlang:halt(Status,[{flush, false}])可以关闭节点呢？我们看它对应的bif代码（erts/emulator/beam/bif.c）：

BIF_RETTYPEhalt_2(BIF_ALIST_2)

{

    …

    if (is_small(BIF_ARG_1) && (code =signed_val(BIF_ARG_1)) >= 0) {

        VERBOSE(DEBUG_SYSTEM,

                ("System halted by BIFhalt(%T, %T)\n", BIF_ARG_1, BIF_ARG_2));

        if (flush) {

            erl_halt((int)(- code));

           ERTS_BIF_YIELD1(bif_export[BIF_halt_1], BIF_P, am_undefined);

        }

        else {

            erts_smp_proc_unlock(BIF_P,ERTS_PROC_LOCK_MAIN);

            erl_exit((int)(-code), "");

        }

}

…

}

如果flush为false，则直接退出节点。