c++ - epoll:丢失一些 EPOLLOUT 事件？

Question

这是我的服务器的样子:

-WorkerThread(s):

• 调用 epoll_wait，接受连接，设置 fd 非阻塞(EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP)
• 在 EPOLLIN 事件上调用 recv 直到 EAGAIN 并将所有数据推送到全局 RecvBuffer(pthread_mutex synced)
• 在 EPOLLOUT 事件上:访问全局 SendBuffer，如果有数据要发送给当前就绪的 fd，则执行此操作(在 while 循环中直到 EAGAIN 或直到所有数据发送完毕；当整个数据包发送完毕时，将其从 SendBuffer 中弹出)<

-IO线程

• 从全局 RecvBuffer 中获取数据，对其进行处理
• 首先尝试立即调用 send 来发送响应。如果没有发送所有数据，将其余数据推送到全局 SendBuffer 以从 WorkerThread 发送)

问题是，服务器不会发送所有排队的数据(它们留在 SendBuffer 中)并且“未发送”的数据量随着客户端数量的增加而增加。为了测试我只使用 1 个工作线程和 1 个 iothread，但如果我使用更多似乎没有任何区别。访问全局缓冲区受 pthread_mutex 保护。此外，我的响应数据大小为 130k 字节(发送此数据量至少需要 3 次发送调用)。另一边是使用阻塞套接字的 Windows 客户端。

非常感谢!兆焦耳

编辑:

是的，默认情况下我正在等待 EPOLLOUT 事件，即使我没有任何东西要发送。为了实现简单性和手册页指南，我是这样做的。另外，我对它的理解是这样的:

即使我当时“错过”了 EPOLLOUT 事件，我不想发送任何东西，这也没问题，因为当我想发送数据时，我会调用发送直到 EAGAIN 和 EPOLLOUT 应该在将来被触发(这是最的时间)

现在我修改了代码以在 IN/OUT 事件之间切换:

关于接受:

event.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
epoll_ctl (pNetServer->m_EventFD, EPOLL_CTL_ADD, infd, &event);

当所有数据发送完毕:

event.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
epoll_ctl (pNetServer->m_EventFD, EPOLL_CTL_MOD, events[i].data.fd, &event);

当我通过在 IOThread 中调用发送到达 EAGAIN 时:

event.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
epoll_ctl (pNetServer->m_EventFD, EPOLL_CTL_MOD, events[i].data.fd, &event);

..我得到了同样的行为。此外，我尝试删除 EPOLLET 标志，但没有任何改变

一个附带的问题:带有 EPOLL_CTL_MOD 标志的 epoll_ctl 是替换事件成员还是只是用给定的参数对它进行或运算？

EDIT3:更新了 IOThread 函数以连续发送，直到发送完所有数据，或直到 EAGAIN。即使我发送了所有数据，我也尝试发送，但大多数时候我在非套接字上收到 errno 88 套接字操作

EDIT4:我修复了“发送代码”中的一些错误，所以我现在没有收到任何未发送的排队数据。但是，我没有收到应有的数据 :))最多的“丢失”(未收到) ) 发送完成后客户端立即调用 recv 时我得到的数据，并且随着客户端数量的增加而增长。当我在客户端的发送和接收调用之间设置 2 秒延迟(阻塞调用)时，我在服务器上丢失的数据很少，这取决于我正在运行的客户端数量(客户端测试代码包括简单的 for 循环，后面有 1 个发送和 1 个接收调用)再次尝试使用和不使用 ET 模式。下面是更新的 WorkerThread 函数，它负责接收数据。@Admins/Mods 也许我现在应该打开新主题，因为问题有点不同？

void CNetServer::WorkerThread(void* param)
{
    CNetServer* pNetServer =(CNetServer*)param;
    struct epoll_event event;
    struct epoll_event *events;
    int s = 0;

//  events = (epoll_event*)calloc (MAXEVENTS, sizeof event);

    while (1)
    {
        int n, i;

//      printf ("BLOCKING NOW! epoll_wait thread %d\n",pthread_self());
        n = pNetServer->m_epollCtrl.Wait(-1);
//      printf ("epoll_wait thread %d\n",pthread_self());
        pthread_mutex_lock(&g_mtx_WorkerThd);
        for (i = 0; i < n; i++)
        {
            if ((pNetServer->m_epollCtrl.Event(i)->events & EPOLLERR))
            {
                // An error has occured on this fd, or the socket is not ready for reading (why were we notified then?)

            //  g_SendBufferArray.RemoveAll( 0 );

                char szFileName[30] = {0};
                sprintf( (char*)szFileName,"fd_%d.txt",pNetServer->m_epollCtrl.Event(i)->data.fd );
                remove(szFileName);

            /*  printf( "\n\n\n");
                printf( "\tDATA LEFT COUNT:%d\n",g_SendBufferArray.size());
                for (int k=0;k<g_SendBufferArray.size();k++)
                    printf( "\tSD: %d DATA LEFT:%d\n",g_SendBufferArray[i]->sd,g_SendBufferArray[i]->nBytesSent );
*/

            //  fprintf (stderr, "epoll error\n");
            //  fflush(stdout);
                close (pNetServer->m_epollCtrl.Event(i)->data.fd);
                continue;
            }
            else if (pNetServer->m_ListenSocket == pNetServer->m_epollCtrl.Event(i)->data.fd)
            {
                // We have a notification on the listening socket, which   means one or more incoming connections. 
                while (1)
                {
                    struct sockaddr in_addr;
                    socklen_t in_len;
                    int infd;
                    char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV];

                    in_len = sizeof in_addr;
                    infd = accept (pNetServer->m_ListenSocket, &in_addr, &in_len);
                    if (infd == -1)
                    {
                        if ((errno == EAGAIN) ||
                            (errno == EWOULDBLOCK))
                        {
                            // We have processed all incoming connections.
                            break;
                        }
                        else
                        {
                            perror ("accept");
                            break;
                        }
                    }

                    s = getnameinfo (&in_addr, in_len,
                        hbuf, sizeof hbuf,
                        sbuf, sizeof sbuf,
                        NI_NUMERICHOST | NI_NUMERICSERV);
                    if (s == 0)
                    {
                        printf("Accepted connection on descriptor %d "
                            "(host=%s, port=%s) thread %d\n", infd, hbuf, sbuf,pthread_self());
                    }

                    // Make the incoming socket non-blocking and add it to the list of fds to monitor.
                    CEpollCtrl::SetNonBlock(infd,true);
                    if ( !pNetServer->m_epollCtrl.Add( infd, EPOLLIN, NULL ))
                    {
                        perror ("epoll_ctl");
                        abort ();
                    }

                }
                continue;
            }
            if( (pNetServer->m_epollCtrl.Event(i)->events & EPOLLOUT) )
            {

                pNetServer->DoSend( pNetServer->m_epollCtrl.Event(i)->data.fd );
            }
            if( pNetServer->m_epollCtrl.Event(i)->events & EPOLLIN )
            {
                printf("EPOLLIN TRIGGERED FOR SD: %d\n",pNetServer->m_epollCtrl.Event(i)->data.fd);
                // We have data on the fd waiting to be read. 
                int done = 0;
                ssize_t count = 0;
                char buf[512];
                while (1)
                {
                    count = read (pNetServer->m_epollCtrl.Event(i)->data.fd, buf, sizeof buf);
                    printf("recv sd %d size %d thread %d\n",pNetServer->m_epollCtrl.Event(i)->data.fd,count,pthread_self());
                    if (count == -1)
                    {
                        // If errno == EAGAIN, that means we have read all data. So go back to the main loop.
                        if ( errno != EAGAIN )
                        {
                            perror ("read");
                            done = 1;
                        }
                        break;
                    }
                    else if (count == 0)
                    {
                        //connection is closed by peer.. do a cleanup and close
                        done = 1;
                        break;
                    }
                    else if (count > 0)
                    {
                        static int nDataCounter = 0;
                        nDataCounter+=count;
                        printf("RECVDDDDD %d\n",nDataCounter);
                        CNetServer::s_pRecvContainer->OnData( pNetServer->m_epollCtrl.Event(i)->data.fd, buf, count );
                    }
                }

                if (done)
                {
                    printf ("Closed connection on descriptor %d\n",pNetServer->m_epollCtrl.Event(i)->data.fd);
                    // Closing the descriptor will make epoll remove it from the set of descriptors which are monitored. 
                    close (pNetServer->m_epollCtrl.Event(i)->data.fd);
                }
            }

        }
//      

        pNetServer->IOThread( (void*)pNetServer );

        pthread_mutex_unlock(&g_mtx_WorkerThd);
    }

}

void CNetServer::IOThread(void* param)
{

    BYTEARRAY* pbPacket = new BYTEARRAY;
    int fd;
    struct epoll_event event;
    CNetServer* pNetServer =(CNetServer*)param;

    printf("IOThread startin' !\n");

    for (;;)
    {
        bool bGotIt = CNetServer::s_pRecvContainer->GetPacket( pbPacket, &fd );

        if( bGotIt )
        {

            //process packet here
            printf("Got 'em packet yo !\n");
            BYTE* re = new BYTE[128000];
            memset((void*)re,0xCC,128000);
            buffer_t* responsebuff = new buffer_t( fd, re, 128000 ) ;

            pthread_mutex_lock(&g_mtx_WorkerThd);

            while( 1 )
            {
                    int s;
                    int nSent = send( responsebuff->sd, ( responsebuff->pbBuffer + responsebuff->nBytesSent ),responsebuff->nSize - responsebuff->nBytesSent,0 );
                    printf ("IOT: Trying to send nSent: %d buffsize: %d \n",nSent,responsebuff->nSize - responsebuff->nBytesSent);

                    if (nSent == -1)
                    {

                        if (errno == EAGAIN || errno == EWOULDBLOCK )
                        {
                                g_vSendBufferArray.push_back( *responsebuff );
                                printf ("IOT: now waiting for EPOLLOUT\n");
                                event.data.fd = fd;
                                event.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP;
                                s = epoll_ctl (pNetServer->m_EventFD, EPOLL_CTL_MOD, fd, &event);
                                break;
                                if (s == -1)
                                {
                                    perror ("epoll_ctl");
                                    abort ();
                                }

                        }
                        else
                        {
                            printf( "%d\n",errno );
                            perror ("send");
                            break;
                        }
                        printf ("IOT: WOOOOT\n");
                        break;
                    }
                    else if (nSent == responsebuff->nSize - responsebuff->nBytesSent)
                    {
                        printf ("IOT:all is sent! wOOhOO\n");
                        responsebuff->sd = 0;
                        responsebuff->nBytesSent += nSent;
                        delete responsebuff;
                        break;
                    }
                    else if (nSent < responsebuff->nSize - responsebuff->nBytesSent)
                    {
                        printf ("IOT: partial send!\n");
                        responsebuff->nBytesSent += nSent;

                    }

            }
            delete [] re;

            pthread_mutex_unlock(&g_mtx_WorkerThd);

        }
    }

}

Answer 1

1. 停止使用 EPOLLET。几乎不可能做到正确。

2. 如果您没有要发送的内容，请不要请求 EPOLLOUT 事件。

3. 当您有数据要通过连接发送时，请遵循以下逻辑:
   
   A) 如果该连接的发送队列中已有数据，只需添加新数据即可。大功告成。
   
   B) 尝试立即发送数据。如果全部发送，则完成。
   
   C) 将剩余的数据保存在该连接的发送队列中。现在请求此连接的 EPOLLOUT。