本文主要是介绍关于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，则直接退出节点。
这篇关于关于erlang socket被动模式和delay_send合用的问题的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
