EPIPE阻止服务器

Question

我在Linux上运行的C中编写了一个单线程异步服务器:套接字是非阻塞的,对于轮询,我使用的是epoll.基准测试显示服务器运行良好,根据Valgrind,没有内存泄漏或其他问题.

唯一的问题是当write()命令被中断时(因为客户端关闭了连接),服务器将遇到EPIPE.我通过使用参数-b运行基准测试实用程序"siege"来人工中断.它连续执行大量请求,这些请求都完美无缺.现在我按CTRL-C并重新启动"围攻".有时我很幸运,服务器无法发送完整的响应,因为客户端的fd无效.正如所料,errno设置为EPIPE.我处理这种情况,在fd上执行close()然后释放与连接相关的内存.现在问题是服务器阻塞并且不再正确回答.这是strace输出:

epoll_wait(4, {{EPOLLIN, {u32=0, u64=0}}}, 128, -1) = 1
accept(3, {sa_family=AF_INET, sin_port=htons(55328), sin_addr=inet_addr("127.0.0.1")}, [16]) = 5
fcntl64(5, F_GETFL)                     = 0x2 (flags O_RDWR)
fcntl64(5, F_SETFL, O_RDWR|O_NONBLOCK)  = 0
epoll_ctl(4, EPOLL_CTL_ADD, 5, {EPOLLIN|EPOLLERR|EPOLLHUP|EPOLLET, {u32=144039912, u64=144039912}}) = 0
epoll_wait(4, {{EPOLLIN, {u32=144039912, u64=144039912}}}, 128, -1) = 1
read(5, "GET /user/register HTTP/1.1\r\nHos"..., 4096) = 161
send(5, "HTTP/1.1 200 OK\r\nContent-Type: t"..., 106, MSG_NOSIGNAL) = 106 <<<<
send(5, "00001000\r\n", 10, MSG_NOSIGNAL) = -1 EPIPE (Broken pipe)        <<<< Why did the previous send() work?
close(5)                                = 0
epoll_wait(4, {{EPOLLIN, {u32=0, u64=0}}}, 128, -1) = 1
accept(3, {sa_family=AF_INET, sin_port=htons(55329), sin_addr=inet_addr("127.0.0.1")}, [16]) = 5
...

(如果你想知道的话,我从日志中删除了printf())

如您所见,客户端建立了一个新的连接,因此被接受.然后,它被添加到EPOLL队列中.epoll_wait()表示客户端发送了数据(EPOLLIN).解析请求并组成响应.发送标题工作正常,但当涉及到正文时,write()会产生EPIPE.它不是"围攻"中的错误,因为它阻止任何传入连接,无论来自哪个客户端.

#include "Connection.h"

static ExceptionManager *exc;

void Connection0(ExceptionManager *e) {
    exc = e;
}

void Connection_Init(Connection *this) {
    Socket_Init(&this->server);
    Socket_SetReusableFlag(&this->server);
    Socket_SetCloexecFlag(&this->server, true);
    Socket_SetBlockingFlag(&this->server, true);
    Socket_ListenTCP(&this->server, 8080, SOMAXCONN);

    // Add the server socket to epoll
    Poll_Init(&this->poll, SOMAXCONN, (void *) &Connection_OnEvent, this);
    Poll_AddEvent(&this->poll, NULL, this->server.fd, EPOLLIN | EPOLLET | EPOLLHUP | EPOLLERR);

    this->activeConn = 0;
}

void Connection_Destroy(Connection *this) {
    Poll_Destroy(&this->poll);
    Socket_Destroy(&this->server);
}

void Connection_Process(Connection *this) {
    Poll_Process(&this->poll, -1);
}

void Connection_AcceptClient(Connection *this) {
    Client *client;

    SocketConnection conn = Socket_Accept(&this->server);
    SocketConnection_SetBlockingFlag(&conn, true);

    client = New(Client);

    client->req = NULL;

    client->conn = New(SocketConnection);
    client->conn->remote = conn.remote;
    client->conn->fd = conn.fd;

    this->activeConn++;

    Poll_AddEvent(&this->poll, client, conn.fd, EPOLLIN | EPOLLET | EPOLLHUP | EPOLLERR);
}

void Connection_DestroyClient(Connection *this, Client *client) {
    // Poll_DeleteEvent(&this->poll, client->conn->fd);
    SocketConnection_Close(client->conn);

    if (client->req != NULL) {
        Request_Destroy(client->req);
        Memory_Free(client->req);
    }

    if (client->conn != NULL) {
        Memory_Free(client->conn);
    }

    Memory_Free(client);

    this->activeConn--;
}

void Connection_OnEvent(Connection *this, int events, Client *client) {
    /* error or connection hung up */
    if (client != NULL && (events & (EPOLLHUP | EPOLLERR))) {
        String_Print(String("EPOLLHUP | EPOLLERR received\n"));
        Connection_DestroyClient(this, client);
        return;
    }

    /* incoming connection */
    if (client == NULL && (events & EPOLLIN)) {
        if (this->activeConn > SOMAXCONN - 1) { /* TODO */
            String_Print(String("Too many connections...\n"));
            return;
        }

        Connection_AcceptClient(this);
        return;
    }

    /* receiving data from client */
    if (client != NULL && (events & EPOLLIN)) {
        if (client->req == NULL) {
            client->req = New(Request);
            Request_Init(client->req, client->conn);
        }

        bool keepOpen = false;

        try (exc) {
            keepOpen = Request_Parse(client->req);
        } catch(&SocketConnection_PipeException, e) {
            printf("Caught PipeException on fd=%d\n", client->conn->fd); fflush(stdout);
        } catch(&SocketConnection_ConnectionResetException, e) {
            printf("Caught ConnectionResetException on fd=%d\n", client->conn->fd); fflush(stdout);
        } finally {
            if (!keepOpen) {
                printf("Will close...\n"); fflush(stdout);
                Connection_DestroyClient(this, client);
            }
        } tryEnd;
    }
}

Answer 1

使用sigaction()设置行动SIGPIPE来SIG_IGN.然后你将获得-1的返回码,errno设置为EPIPE,没有信号.

在Linux上,另一种是利用send()与MSG_NOSIGNAL标志代替,write().这允许您抑制该写入的信号,而不会影响任何其他写入.BSD系统的另一种选择是SO_NOSIGPIPE套接字选项,它可以抑制SIGPIPE该套接字上的所有写操作.

在任何时候,内核TCP实现都可以从对等端接收TCP RST.在该点之后的套接字上的下一次写入将导致EPIPE错误,并且SIGPIPE如果未被抑制则将发出信号.因此,即使连续两次写入,第一次写入也可能成功,下一次写入可能会因EPIPE和SIGPIPE而失败.

更新:虽然它不是发布的代码的一部分,但我在您的strace输出中看到您正在调用fcntl64(5, F_SETFL, O_RDONLY|O_NONBLOCK).因为O_RDONLY是0,可能你的代码只是将标志设置为O_NONBLOCK.它应该获取当前标志然后将其OldFlags | O_NONBLOCK设置为以设置非阻塞模式.将套接字设置为只读模式似乎可能在写入时导致问题.或者可能是你不小心使用F_GETFD而不是F_GETFL获得旧标志.