使用kqueue异步发送数据

Question

我有一个用普通C编写的服务器，它在FreeBSD上使用kqueue接受TCP连接。

接收传入的连接，并将其添加到一个简单的连接池中，以跟踪文件句柄。

当接收到数据(在EVFILT_READ上)时，我调用recv()，然后将有效负载放在一个消息队列中，以供不同的线程处理。这种方式对数据的接收和处理是非常完美的。

当处理线程完成时，它可能需要向客户端发送一些东西。由于处理线程可以访问连接池，并且可以很容易地获得文件句柄，所以我只是从处理线程调用send()。

这在99%的情况下都能工作，但是kqueue不时地给我一个EV_EOF标志，然后删除连接。

发送()调用的频率与EV_EOF错误的数量之间存在明显的相关性，因此，由于kqueue线程与处理线程之间的某些争用条件，我感觉到了EV_EOF。

对send()的调用总是返回预期的字节计数，因此我不会填充tx缓冲区。

所以是我的问题；是否可以像这里描述的那样从一个单独的线程调用send()？如果不是，异步地将数据发送回客户端的正确方法是什么？

我发现调用的所有示例都在与kqueue循环相同的上下文中发送()，但是我的处理线程可能需要在任何时候--甚至在上次从客户端接收数据之后几分钟--发送回数据，因此很明显，我不能阻止那个时候的kqueue循环。

相关代码片段：

void    *tcp_srvthread(void *arg)
{
    [[...Bunch of declarations...]]

    tcp_serversocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);        
    ...
    setsockopt(tcp_serversocket, SOL_SOCKET, SO_REUSEADDR, &i, sizeof(int));            
    ...
    err = bind(tcp_serversocket, (const struct sockaddr*)&sa, sizeof(sa));
    ...
    err = listen(tcp_serversocket, 10);
    ...
    kq = kqueue();    
    EV_SET(&evSet, tcp_serversocket, EVFILT_READ | EV_CLEAR, EV_ADD, 0, 0, NULL);
    ...
    while(!fTerminated) {
        timeout.tv_sec = 2;  timeout.tv_nsec = 0;
        nev = kevent(kq, &evSet, 0, evList, NLIST, &timeout);

        for (i=0; i<nev; i++) {            
            if (evList[i].ident == tcp_serversocket) {                              // new connection?
                socklen = sizeof(addr);
                fd = accept(evList[i].ident, &addr, &socklen);   // accept it
                if(fd > 0) {                                                       // accept ok?
                    uidx = conn_add(fd, (struct sockaddr_in *)&addr);               // Add it to connected controllers
                    if(uidx >= 0) {                                                 // add ok?                        

                        EV_SET(&evSet, fd, EVFILT_READ | EV_CLEAR, EV_ADD, 0, 0, (void*)(uint64_t)(0x00E20000 | uidx));        // monitor events from it
                        if (kevent(kq, &evSet, 1, NULL, 0, NULL) == -1) {           // monitor ok?
                            conn_delete(uidx);                                      // ..no, so delete it from my list also
                        }
                    } else {                                                        // no room on server?
                        close(fd);
                    }
                }
                else Log(0, "ERR: accept fd=%d", fd);
            }
            else
            if (evList[i].flags & EV_EOF) {
                [[ ** THIS IS CALLED SOMETIMES AFTER CALLING SEND - WHY??  ** ]]
                uidx = (uint32_t)evList[i].udata;                
                conn_delete( uidx );               
            }
            else
            if (evList[i].filter == EVFILT_READ) {                  
                if((nr = recv(evList[i].ident, buf, sizeof(buf)-2, 0)) >     0) {
                    uidx = (uint32_t)evList[i].udata;
                    recv_data(uidx, buf, nr);    // This will queue the message for the processing thread
                }
            }
        }
        else {
            // should not get here.
        }

    }
}

处理线程看起来如下(显然，除了显示的内容之外，还有很多数据操作)：

void    *parsethread(void *arg)
{         
    int                 len;
    tmsg_Queue          mq;
    char                is_ok;

    while(!fTerminated) {
        if((len = msgrcv(msgRxQ, &mq, sizeof(tmsg_Queue), 0, 0)) > 0) {            
             if( process_message(mq) ) {
                 [[ processing will find the uidx of the client and build the return data ]]
                 send( ctl[uidx].fd, replydata, replydataLen, 0 );
             }
        }
    }
}

欣赏任何正确的想法或建议。谢谢。

Answer 1

EV_EOF

如果您在对等方关闭其读取部分后写入套接字，您将收到一个RST，这将触发具有EVFILT_READ集的EV_EOF。

异步

你应该试试aio_read和aio_write。