前段时间有同学在线上问了个问题：

服务器端我是这样设的：gen_tcp:listen(8000, [{active, false}, {recbuf,1}, {buffer,1}]).

客户端是这样设的：gen_tcp:connect(“localhost”, 8000, [{active, false}， {high_watermark,2}, {low_watermark,1}, {sndbuf,1}, {buffer,1}]).

我客户端每次gen_tcp:send()发送一个字节，前6个字节返回ok，第7个字节阻塞

服务端每次gen_tcp:recv(_,0)接收一个字节，接收三个字节后，客户端的第7次发送返回。

按我的理解的话：应该是 服务器端可以接收2个字节+sndbuf里的一个字节，第4个字节客户端就该阻塞的，可事实不时这样，求分析

这个问题确实还是比较复杂，涉及到gen_tcp的发送缓冲区和接收缓冲区，水位线等问题，其中接收缓冲区的问题在这篇 以及这篇 博文里面讲的比较清楚了，今天我们重点来分析下发送缓冲区和水位线的问题。

在开始分析前，我们需要熟悉几个gen_tcp的选项, 更多参见 这里：

{delay_send, Boolean}

Normally, when an Erlang process sends to a socket, the driver will try to immediately send the data. If that fails, the driver will use any means available to queue up the message to be sent whenever the operating system says it can handle it. Setting {delay_send, true} will make all messages queue up. This makes the messages actually sent onto the network be larger but fewer. The option actually affects the scheduling of send requests versus Erlang processes instead of changing any real property of the socket. Needless to say it is an implementation specific option. Default is false.

{high_msgq_watermark, Size} (TCP/IP sockets)

The socket message queue will be set into a busy state when the amount of data queued on the message queue reaches this limit. Note that this limit only concerns data that have not yet reached the ERTS internal socket implementation. Default value used is 8 kB.

Senders of data to the socket will be suspended if either the socket message queue is busy, or the socket itself is busy.

For more information see the low_msgq_watermark, high_watermark, and low_watermark options.

Note that distribution sockets will disable the use of high_msgq_watermark and low_msgq_watermark, and will instead use the distribution buffer busy limit which is a similar feature.

{high_watermark, Size} (TCP/IP sockets)

The socket will be set into a busy state when the amount of data queued internally by the ERTS socket implementation reaches this limit. Default value used is 8 kB.

Senders of data to the socket will be suspended if either the socket message queue is busy, or the socket itself is busy.

For more information see the low_watermark, high_msgq_watermark, and low_msqg_watermark options.

{low_msgq_watermark, Size} (TCP/IP sockets)

If the socket message queue is in a busy state, the socket message queue will be set in a not busy state when the amount of data queued in the message queue falls below this limit. Note that this limit only concerns data that have not yet reached the ERTS internal socket implementation. Default value used is 4 kB.

Senders that have been suspended due to either a busy message queue or a busy socket, will be resumed when neither the socket message queue, nor the socket are busy.

For more information see the high_msgq_watermark, high_watermark, and low_watermark options.

Note that distribution sockets will disable the use of high_msgq_watermark and low_msgq_watermark, and will instead use the distribution buffer busy limit which is a similar feature.

{low_watermark, Size} (TCP/IP sockets)

If the socket is in a busy state, the socket will be set in a not busy state when the amount of data queued internally by the ERTS socket implementation falls below this limit. Default value used is 4 kB.

Senders that have been suspended due to either a busy message queue or a busy socket, will be resumed when neither the socket message queue, nor the socket are busy.

For more information see the high_watermark, high_msgq_watermark, and low_msgq_watermark options.

两对高低水位线的设置，以及delay_send选项，对发送缓冲区的影响很大。

gen_tcp:send的行为在之前的 博文 中分析的比较到位了，建议同学先看看这篇文章。

我们知道每个erlang的进程都有个消息队列，其他进程要和他通信就需要透过发消息给他，把通讯的内容在消息里面交代清楚。进程消息队列里面一旦有消息，erlang的VM就会马上准备调度该进程来让进程执行，处理消息。这个进程的消息队列机制每个erlang入门的书籍都写的非常清楚。 那么port呢？在Erlang的早期，Port是和进程一样的地位，接口，使用方式。Port作为Erlang对外的IO的执行单位，也拥有自己的消息队列，当进程把消息发送给port的时候，port通常也是把消息保存在消息队列中，然后VM就会调度这个port。等到port被调度执行的时候，port把队列里面的消息消耗掉，发送到网络或者执行相应IO的操作。port的调度和erlang的进程的调度是一样的，都非常讲究公平调度。

我们来考证下port和进程消息发送的接口。 我们知道!符号是erlang:send的语法糖，当我们给Port!msg 或者Pid!msg，最终都是调用erlang:send来发送消息。后面不知道为什么，erlang的设计者专门为port设计了port_command系列函数专门为port发送消息。

我们来考证下：

erlang:send->BIF_RETTYPE send_3(BIF_ALIST_3)->do_send  源码在bif.c中我们来看看：

Sint
do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
...
    if (is_internal_pid(to)) {
    ...
  
    } else if (is_external_pid(to)) {
    ...
    return remote_send(p, dep, to, to, msg, suspend);
    } else if (is_atom(to)) {
    ...
    } else if (is_external_port(to)
               && (external_port_dist_entry(to)
                   == erts_this_dist_entry)) {
        erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
        erts_dsprintf(dsbufp,
                      "Discarding message %T from %T to %T in an old "
                      "incarnation (%d) of this node (%d)\n",
                      msg,
                      p->common.id,
                      to,
                      external_port_creation(to),
                      erts_this_node->creation);
        erts_send_error_to_logger(p->group_leader, dsbufp);
        return 0;
    } else if (is_internal_port(to)) {
    ...
        pt = erts_port_lookup(portid, ERTS_PORT_SFLGS_INVALID_LOOKUP);
        ...
            switch (erts_port_command(p, ps_flags, pt, msg, refp)) {
            case ERTS_PORT_OP_CALLER_EXIT:
...
}

诸位看到了吧！ 1. erlang:send接受二种对象： port和process 2. 发送到port的消息走的和erts_port_command是一样的路。

喝口水，保存体力，重新温习下二点： 1. port有消息队列。 2. port也是公平调度。

有了上面的知识铺垫，我们其实就比较好明白上面选项中的水位线做什么的。和每个消息队列一样，为了防止发送者和接收者能力的失衡，通常都会设置高低水位线来保护队列不至于太大把

系统撑爆。 上面的{high_watermark, Size}，{low_watermark, Size} 就是干这个用的。

那port是如何保护自己的呢？答案是：

当消息量达到高水位线的时候，port进入busy状态，这时候会把发送进程suspend起来，等消息达到低水位线的时候，解除busy状态，同时让发送进程继续执行。

证明上面的说法，参考下port_command 文档：

port_command(Port, Data, OptionList) -> boolean()

Types:

Port = port() | atom()

Data = iodata()

Option = force | nosuspend

OptionList = [Option]

Sends data to a port. port_command(Port, Data, []) equals port_command(Port, Data).

If the port command is aborted false is returned; otherwise, true is returned.

If the port is busy, the calling process will be suspended until the port is not busy anymore.

Currently the following Options are valid:

force

The calling process will not be suspended if the port is busy; instead, the port command is forced through. The call will fail with a notsup exception if the driver of

the port does not support this. For more information see the ERL_DRV_FLAG_SOFT_BUSY driver flag.

nosuspend

The calling process will not be suspended if the port is busy; instead, the port command is aborted and false is returned.

那如何知道一个port进入busy状态，因为这个状态通常很严重，发送进程被挂起，会引起很大的latency.

幸亏erlang考虑周到，参看这里：

erlang:system_monitor(MonitorPid, Options) -> MonSettings

busy_port

If a process in the system gets suspended because it sends to a busy port, a message {monitor, SusPid, busy_port, Port} is sent to MonitorPid. SusPid is the pid that

got suspended when sending to Port.

系统会很友好的把发生busy_port的进程发出来，我们就可以知道那个进程进程碰到高水位线被挂起了，方面我们后面调整水位线避免这种情况发生。

当用户调用gen_tcp:send要发送数据的时候最终都会调用port_command来具体执行, 那么我们来看下它是如何运作的：

/* Command should be of the form                                                                                          
**   {PID, close}                                                                                                         
**   {PID, {command, io-list}}                                                                                            
**   {PID, {connect, New_PID}}                                                                                            
*/
ErtsPortOpResult
erts_port_command(Process *c_p,
                  int flags,
                  Port *port,
                  Eterm command,
                  Eterm *refp)
{
...
   if (is_tuple_arity(command, 2)) {
        Eterm cntd;
        tp = tuple_val(command);
        cntd = tp[1];
        if (is_internal_pid(cntd)) {
            if (tp[2] == am_close) {
                if (!erts_port_synchronous_ops)
                    refp = NULL;
                flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
                return erts_port_exit(c_p, flags, port, cntd, am_normal, refp);
            } else if (is_tuple_arity(tp[2], 2)) {
                tp = tuple_val(tp[2]);
                if (tp[1] == am_command) {
                    if (!(flags & ERTS_PORT_SIG_FLG_NOSUSPEND)
                        && !erts_port_synchronous_ops)
                    refp = NULL;
                    return erts_port_output(c_p, flags, port, cntd, tp[2], refp);
                }
                else if (tp[1] == am_connect) {
                    if (!erts_port_synchronous_ops)
                        refp = NULL;
                    flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND;
                    return erts_port_connect(c_p, flags, port, cntd, tp[2], refp);
                }
            }
        }
    }
}
...
}
  
ErtsPortOpResult
erts_port_output(Process *c_p,
                 int flags,
                 Port *prt,
                 Eterm from,
                 Eterm list,
                 Eterm *refp)
{
...
    try_call = (force_immediate_call /* crash dumping */
                || !(sched_flags & (invalid_flags
                                    | ERTS_PTS_FLGS_FORCE_SCHEDULE_OP)));
  
    if (drv->outputv) {
          try_call_state.pre_chk_sched_flags = 0; /* already checked */
            if (force_immediate_call)
                try_call_res = force_imm_drv_call(&try_call_state);
            else
                try_call_res = try_imm_drv_call(&try_call_state);
            switch (try_call_res) {
            case ERTS_TRY_IMM_DRV_CALL_OK:
                call_driver_outputv(flags & ERTS_PORT_SIG_FLG_BANG_OP,
                                    c_p ? c_p->common.id : ERTS_INVALID_PID,
                                    from,
                                    prt,
                                    drv,
                                    evp);
                if (force_immediate_call)
                    finalize_force_imm_drv_call(&try_call_state);
                else
                    finalize_imm_drv_call(&try_call_state);
                /* Fall through... */
...
}
  
static ERTS_INLINE void
call_driver_outputv(int bang_op,
                    Eterm caller,
                    Eterm from,
                    Port *prt,
                    erts_driver_t *drv,
                    ErlIOVec *evp)
{
    /*                                                                                                                    
     * if (bang_op)                                                                                                       
     *   we are part of a "Prt ! {From, {command, Data}}" operation                                                       
     * else                                                                                                               
     *   we are part of a call to port_command/[2,3]                                                                      
     * behave accordingly...                                                                                              
     */
    if (bang_op && from != ERTS_PORT_GET_CONNECTED(prt))
        send_badsig(prt);
    else {
...
        prt->caller = caller;
        (*drv->outputv)((ErlDrvData) prt->drv_data, evp);
        prt->caller = NIL;
  
        prt->bytes_out += size;
        erts_smp_atomic_add_nob(&erts_bytes_out, size);
    }
...
}

从源码分析来看，我们看到port_command如果看到port要执行command命令就会调用erts_port_output， 而后者会做复杂的判断，来决定如何调用call_driver_outputv。

这个复杂的流程就是msgq_watermark水位线发挥作用地方，我们暂时不分析，等后面讲msgq_watermark的时候一起。

目前只需要知道最终gen_tcp:send发松数据会调用port driver的outputv回调函数输出就好了。

接着源码分析:

static struct erl_drv_entry tcp_inet_driver_entry =
{
    tcp_inet_init,  /* inet_init will add this driver !! */
    tcp_inet_start,
    tcp_inet_stop,
    tcp_inet_command,
    tcp_inet_drv_input,
    tcp_inet_drv_output,
    "tcp_inet",
    NULL,
    NULL,
    tcp_inet_ctl,
    tcp_inet_timeout,
    tcp_inet_commandv,
...
}
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;
            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));
}

对于inet_drv(gen_tcp)的例子来讲就是会调用tcp_sendv来把消息转变成网络封包发送出去。

好吧，喝口水，休息下。 这里我们梳理下我们的数据路线：

gen_tcp:send->port_command->erts_port_output->call_driver_outputv->tcp_inet_commandv->tcp_sendv

大家要牢记在心。

继续接着我们参照源码来分析下水位线的实现：

/* inet_drv.c */
#define INET_LOPT_TCP_HIWTRMRK     27  /* set local high watermark */
#define INET_LOPT_TCP_LOWTRMRK     28  /* set local low watermark */
  
  
#define INET_HIGH_WATERMARK (1024*8) /* 8k pending high => busy  */
#define INET_LOW_WATERMARK  (1024*4) /* 4k pending => allow more */
  
typedef struct {
...
    int   high;                 /* high watermark */
    int   low;                  /* low watermark */
...
} tcp_descriptor;
  
static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
{
...
    desc->high = INET_HIGH_WATERMARK;
    desc->low  = INET_LOW_WATERMARK;
...
}
  
static int inet_set_opts(inet_descriptor* desc, char* ptr, int len)
{
...
        case INET_LOPT_TCP_HIWTRMRK:
            if (desc->stype == SOCK_STREAM) {
                tcp_descriptor* tdesc = (tcp_descriptor*) desc;
                if (ival < 0) ival = 0;
                if (tdesc->low > ival)
                    tdesc->low = ival;
                tdesc->high = ival;
            }
            continue;
  
        case INET_LOPT_TCP_LOWTRMRK:
            if (desc->stype == SOCK_STREAM) {
                tcp_descriptor* tdesc = (tcp_descriptor*) desc;
                if (ival < 0) ival = 0;
                if (tdesc->high < ival)
                    tdesc->high = ival;
                tdesc->low = ival;
            }
            continue;
...
}
  
/* Copy a descriptor, by creating a new port with same settings                                                           
 * as the descriptor desc.                                                                                                
 * return NULL on error (SYSTEM_LIMIT no ports avail)                                                                     
 */
static tcp_descriptor* tcp_inet_copy(tcp_descriptor* desc,SOCKET s,
                                     ErlDrvTermData owner, int* err)
{
...
    copy_desc->high          = desc->high;
    copy_desc->low           = desc->low;
...
}
  
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;
        }
...
}
  
/* socket ready for ouput:                                                                                                
** 1. INET_STATE_CONNECTING => non block connect ?                                                                        
** 2. INET_STATE_CONNECTED  => write output                                                                               
*/
static int tcp_inet_output(tcp_descriptor* desc, HANDLE event)
{
...
    if (driver_deq(ix, n) <= desc->low) {
                if (IS_BUSY(INETP(desc))) {
                    desc->inet.caller = desc->inet.busy_caller;
                    desc->inet.state &= ~INET_F_BUSY;
                    set_busy_port(desc->inet.port, 0);
                    /* if we have a timer then cancel and send ok to client */
                    if (desc->busy_on_send) {
                        driver_cancel_timer(desc->inet.port);
                        desc->busy_on_send = 0;
                    }
                    inet_reply_ok(INETP(desc));
                }
            }
...
}

从源码我们可以分析出几点：

1. 水位线设置是可以继承的。

2. 高低水位线默认是8K/4K.

3. 进入高水位后，port进入busy状态。

4. 当消息消耗到小于低水位线，busy解除。

这个水位线的说明和文档解释的一样，接下来我们稍微看看delay_send的实现原理,还是继续上源码：

/* TCP additional flags */
#define TCP_ADDF_DELAY_SEND    1
  
static int inet_set_opts(inet_descriptor* desc, char* ptr, int len)
{
...
case INET_LOPT_TCP_DELAY_SEND:
            if (desc->stype == SOCK_STREAM) {
                tcp_descriptor* tdesc = (tcp_descriptor*) desc;
                if (ival)
                    tdesc->tcp_add_flags |= TCP_ADDF_DELAY_SEND;
                else
                    tdesc->tcp_add_flags &= ~TCP_ADDF_DELAY_SEND;
            }
            continue;
...
}
  
/*                                                                                                                        
** Send non-blocking vector data                                                                                          
*/
static int tcp_sendv(tcp_descriptor* desc, ErlIOVec* ev)
{
...
        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);
            }
            n = 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, Send failed, 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);
...
}
static void tcp_inet_drv_output(ErlDrvData data, ErlDrvEvent event)
{
    (void)tcp_inet_output((tcp_descriptor*)data, (HANDLE)event);
}
/* socket ready for ouput:                                                                                                
** 1. INET_STATE_CONNECTING => non block connect ?                                                                        
** 2. INET_STATE_CONNECTED  => write output                                                                               
*/
static int tcp_inet_output(tcp_descriptor* desc, HANDLE event)
{
...
 else if (IS_CONNECTED(INETP(desc))) {
        for (;;) {
            int vsize;
            ssize_t 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;
                }
            goto done;
            }
            if (driver_deq(ix, n) <= desc->low) {
                if (IS_BUSY(INETP(desc))) {
                    desc->inet.caller = desc->inet.busy_caller;
                    desc->inet.state &= ~INET_F_BUSY;
                    set_busy_port(desc->inet.port, 0);
                    /* if we have a timer then cancel and send ok to client */
                    if (desc->busy_on_send) {
                        driver_cancel_timer(desc->inet.port);
                        desc->busy_on_send = 0;
                    }
             inet_reply_ok(INETP(desc));
                }
            }
...
}

从源码分析我们可以知道当tcp_sendv发送数据前看下：

1. delay_send标志是否设置，如果设置就不尝试调用sock_sendv发送。

2. 调用sock_sendv发送网络数据，剩下的部分数据保存到驱动的队列去。

3. 如果队列有数据的话，就把把epoll的写事件挂载上。

4. 后续epoll会通知socket可写的时候，会调用tcp_inet_drv_output

5. tcp_inet_drv_output->tcp_inet_output 继续把之前在队列里面的数据透过sock_sendv再次发送到网络

步骤3和4之间需要时间，依赖于epoll的写事件发生的以及port调度的时间点。

所以简单的说： delay_send就是在第一阶段不尝试发送数据，直接把数据推入port的消息队列去，等后面epoll说socket可写的时候一起发送出去。

这种做法的好处是gen_tcp:send马上就可以返回，因为sock_send通常要耗费几十us的时间，可用在对发送的latency很敏感的场合。

到这里为止，我们清楚的分析了数据是如何在port的各个链条里面流动.

再回顾下：当gen_tcp:send数据无法离开通过网络发送出去的时候，会暂时保留在port的消息队列里面，当消息队列满(到高水位线)的时候，port就会busy，抑制发送者推送更多的数据。当epoll探测到socket可写的时候，vm会调用tcp_inet_output把消息队列里面的数据，拉到网络去，这个过程中，队列里面的数据会越来越少，少到低水位线的时候，解除busy, 好让发送者发送更多的数据。

再喝口水，我们接着分析msgq_watermark.

待续！！！

总结： 这个水位线官方文档写的不清不楚，还是源码靠谱！

建议继续学习：

1. gen_tcp发送进程被挂起起因分析及对策    (阅读：36909)
2. gen_tcp调用进程收到{empty_out_q, Port}消息奇怪行为分析    (阅读：3485)
3. gen_tcp容易误用的一点解释    (阅读：2572)
4. gen_tcp如何限制封包大小    (阅读：2363)
5. pdflush 相关    (阅读：2300)
6. 思考mysql内核之初级系列4--innodb缓冲区管理    (阅读：2305)
7. RAID卡MTRR的RAID模式write-combining    (阅读：2027)
8. 未公开的gen_tcp:unrecv以及接收缓冲区行为分析    (阅读：1885)
9. gen_tcp接受链接时enfile的问题分析及解决    (阅读：1580)
10. 未公开的gen_tcp:unrecv以及接收缓冲区行为分析    (阅读：1387)