c++ - 第一个包无法使用非阻塞 tcp 套接字发送

Question

我正在编写一个使用TCP协议(protocol)的工具。该工具模拟1000个TCP客户端，当socket fd增加到500个左右时，后续创建的socket无法发送数据，send函数返回EAGAIN。为什么？

我所有的 TCP 套接字都是非阻塞的。

例如:

我的测试日志:

12 2012-12-03 23:21:56:41005 tang_dts_serv_task.cpp:55 [3] [17294] dts zero channel acc.[confid:41075200][userid:860194970]
 13 2012-12-03 23:21:56:41509 send_task_mgr.cpp:49 [3] [17348] send task mgr.[cmdcode:4354][fd:1026]
 14 2012-12-03 23:21:56:41529 data_send_task.cpp:51 [2] [17348] dts send task.[cmdcode:4354][fd:1026][port:25789]
 15 2012-12-03 23:21:56:41543 os_sock.cpp:15 [0] [17348] send error.[fd:1026][errno:11]

socket 1026刚刚创建，但是第一个包无法发送。

我的经历:

Ubuntu 8.04.4 LTS
Linux VM-Ubuntu203001 2.6.24-24-server #1 SMP Fri Sep 18 17:24:10 UTC 2009 i686 GNU/Linux

发送部分:

void DataSendTask::ProcessMsg(message *msg)
{
    tcm_message *tm;
    DtsContext *ctx;
    ClientEntity *cli;

    size_t transfer_bytes;
    char *d_buf;
    int buf_len;
    int retval;
    EventHandler *handler;
    Reactor *reactor;

    tm = (tcm_message*)msg->content;
    ctx = (DtsContext*)tm->ctx;

    handler = ctx->event_handler();
    reactor = handler->reactor();

    SendBuffer &s_buf = ctx->send_buf();
    SockStream &stream = ctx->sock_stream(); 
    cli = ctx->client_entity();

    int sock_state = ctx->GetSockState();
    if (sock_state != DtsContext::CS_CONNECTED){
        LOG_ERROR("data send task socket state exception! [fd:%d][sockstate:%d]" ,stream.sock_fd() , sock_state);
        delete tm;
        tm = NULL;

        return;
    }

    struct sockaddr_in localaddr;
    socklen_t addr_len;
    addr_len = sizeof(localaddr);

    OsSock::GetSockName(stream.sock_fd() , (struct sockaddr*)&localaddr , &addr_len);
    int port = localaddr.sin_port;
    LOG_INFO("dts send task.[cmdcode:%d][fd:%d][port:%d]",msg->msg_code , stream.sock_fd() , port);

    //1. Wei need to see wheter the half_package have unsend bytes.
    buf_len = s_buf.GetHalfPackageLength();

    if (buf_len > 0){
        transfer_bytes = 0;
        d_buf = s_buf.GetHalfPackagePointer();
        retval = stream.Send(d_buf , buf_len , MSG_NOSIGNAL , &transfer_bytes); 
        if (retval <= 0){
            LOG_ERROR("dts half send error.[fd:%d][send_bytes:%d][transfered_bytes:%d]" ,stream.sock_fd(),buf_len,transfer_bytes);
            if (retval == OS_SOCK_EAGAIN){
                //register event handler
                reactor->RegisterHandler(handler , EventHandler::EM_WRITE);
            }else if (retval == OS_SOCK_PEER_CNN_CLOSED){
                //connect again?
                ClientMgr::Instance()->ClientOffline(cli);
            }else{
                //error,print log info,continue;
            }

            s_buf.SaveLeftData(d_buf-transfer_bytes , buf_len - transfer_bytes);
            delete tm;
            tm = NULL;

            return;
        }

        s_buf.SetHalfPackageLength(0);

        LOG_DEBUG("dts half send success.[bytes:%d][fd:%d]",retval , stream.sock_fd());
    }

    PriorityQueue<DtsRtpPackage> &squeue = ctx->send_queue();
    while(1){
        //1. Get message from priority queue.
        DtsRtpPackage *data = squeue.Dequeue();

        if (data == NULL)
            break;

        d_buf = (char *)data; 
        buf_len = data->m_size + 4;

        transfer_bytes = 0;
        retval = stream.Send(d_buf , buf_len , MSG_NOSIGNAL  , &transfer_bytes); 
        if (retval <= 0){
            LOG_ERROR("dts send error.[fd:%d][send_bytes:%d][transfered_bytes:%d]" ,stream.sock_fd(),buf_len,transfer_bytes);

            if (retval == OS_SOCK_EAGAIN){
                //register event handler
                reactor->RegisterHandler(handler , EventHandler::EM_WRITE);
            }else if (retval == OS_SOCK_PEER_CNN_CLOSED){
                //connect again?
                ClientMgr::Instance()->ClientOffline(cli);
            }else{
                //error,print log info,continue;
            }

            s_buf.SaveLeftData(d_buf-transfer_bytes , buf_len - transfer_bytes);
            tang_mfree(data);

            break;
        }
        LOG_DEBUG("dts send success.[bytes:%d][fd:%d]",retval , stream.sock_fd());

        tang_mfree(data);
    }

    delete tm;
    tm = NULL;

    LOG_DEBUG("dts send over.[fd:%d]", stream.sock_fd());
}

Answer 1

当您使用非阻塞套接字时，您必须预料到有时您会在必须阻塞的时候尝试执行某个操作。这就是非阻塞套接字的本质。当它们本来会被阻止时，它们会返回错误。

您的代码如下:

    retval = stream.Send(d_buf , buf_len , MSG_NOSIGNAL , &transfer_bytes); 
    if (retval <= 0){
        LOG_ERROR("dts half send error.[fd:%d][send_bytes:%d][transfered_bytes:%d]" ,stream.sock_fd(),buf_len,transfer_bytes);
        if (retval == OS_SOCK_EAGAIN){ // <- HERE
            //register event handler
            reactor->RegisterHandler(handler , EventHandler::EM_WRITE);
        }else if (retval == OS_SOCK_PEER_CNN_CLOSED){
            //connect again?
            ClientMgr::Instance()->ClientOffline(cli);
        }else{
            //error,print log info,continue;
        }

因此它已经正确处理了这种情况，并在有可用缓冲区空间时重试写入。我只需更改代码以不记录 OS_SOCK_EAGAIN 或 OS_SOCK_PEER_CNN_CLOSED，因为这些实际上并不是错误，而是正常的预期条件。