`
810364804
  • 浏览: 784208 次
文章分类
社区版块
存档分类
最新评论

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

 
阅读更多

分类:Erlang622人阅读评论(0)收藏举报

转载请注明,来自: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,则直接退出节点。

转载请注明,来自: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,则直接退出节点。

分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics