zoukankan      html  css  js  c++  java
  • 关于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送数据会出现closedENOTCONN错误;同样如果没设置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都这么有加有减的话,关闭完所有porterts_halt_progress值还为1,接着将erts_halt_progress1,若结果为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,则直接退出节点。


  • 相关阅读:
    java synchronized
    [多问几个为什么]为什么匿名内部类中引用的局部变量和参数需要final而成员字段不用?(转)
    Java中static、final用法小结(转)
    使用 Spring 2.5 注释驱动的 IoC 功能(转)
    (转载)Spring 注解@Component,@Service,@Controller,@Repository
    Spring中@Autowired注解、@Resource注解的区别
    控制反转和依赖注入模式(转)
    HDU 4758 Walk Through Squares (2013南京网络赛1011题,AC自动机+DP)
    HDU 4745 Two Rabbits (2013杭州网络赛1008,最长回文子串)
    HDU 4747 Mex (2013杭州网络赛1010题,线段树)
  • 原文地址:https://www.cnblogs.com/javawebsoa/p/2990487.html
Copyright © 2011-2022 走看看