Reactor模式服务器无连接提示、不返回响应及Ctrl+C断开问题排查

1. reactor.h（核心头文件） #

#ifndef REACTOR_H
#define REACTOR_H

#include <sys/epoll.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>

// 事件处理函数类型定义
typedef void (*event_handler_t)(int fd, int events, void *arg);

// 反应堆中的事件节点
typedef struct reactor_event {
    int fd;
    int events; // EPOLLIN/EPOLLOUT等
    event_handler_t handler;
    void *arg;
    struct reactor_event *next;
} reactor_event_t;

// 反应堆结构体
typedef struct reactor {
    int epfd; // epoll实例描述符
    struct epoll_event *events; // 存储就绪事件的数组
    int max_events; // 最大处理事件数
    reactor_event_t *event_list; // 注册的事件链表
} reactor_t;

// 反应堆操作函数声明
reactor_t *reactor_init(int max_events);
int reactor_add_event(reactor_t *reactor, int fd, int events, event_handler_t handler, void *arg);
int reactor_remove_event(reactor_t *reactor, int fd);
void reactor_run(reactor_t *reactor);
void reactor_destroy(reactor_t *reactor);

#endif // REACTOR_H

2. reactor.c（反应堆核心实现） #

#include "reactor.h"

reactor_t *reactor_init(int max_events) {
    reactor_t *reactor = (reactor_t *)malloc(sizeof(reactor_t));
    if (!reactor) {
        perror("malloc reactor failed");
        return NULL;
    }

    reactor->epfd = epoll_create1(0);
    if (reactor->epfd == -1) {
        perror("epoll_create1 failed");
        free(reactor);
        return NULL;
    }

    reactor->max_events = max_events;
    reactor->events = (struct epoll_event *)malloc(sizeof(struct epoll_event) * max_events);
    if (!reactor->events) {
        perror("malloc epoll events failed");
        close(reactor->epfd);
        free(reactor);
        return NULL;
    }

    reactor->event_list = NULL;
    return reactor;
}

int reactor_add_event(reactor_t *reactor, int fd, int events, event_handler_t handler, void *arg) {
    struct epoll_event ev;
    memset(&ev, 0, sizeof(ev));
    ev.events = events;
    ev.data.fd = fd;

    if (epoll_ctl(reactor->epfd, EPOLL_CTL_ADD, fd, &ev) == -1) {
        perror("epoll_ctl add failed");
        return -1;
    }

    // 创建事件节点并添加到链表
    reactor_event_t *event = (reactor_event_t *)malloc(sizeof(reactor_event_t));
    if (!event) {
        perror("malloc event node failed");
        epoll_ctl(reactor->epfd, EPOLL_CTL_DEL, fd, NULL);
        return -1;
    }
    event->fd = fd;
    event->events = events;
    event->handler = handler;
    event->arg = arg;
    event->next = reactor->event_list;
    reactor->event_list = event;

    return 0;
}

int reactor_remove_event(reactor_t *reactor, int fd) {
    if (epoll_ctl(reactor->epfd, EPOLL_CTL_DEL, fd, NULL) == -1) {
        perror("epoll_ctl del failed");
        return -1;
    }

    // 从链表中移除事件节点
    reactor_event_t **curr = &reactor->event_list;
    while (*curr) {
        if ((*curr)->fd == fd) {
            reactor_event_t *temp = *curr;
            *curr = temp->next;
            free(temp);
            break;
        }
        curr = &(*curr)->next;
    }

    close(fd);
    return 0;
}

void reactor_run(reactor_t *reactor) {
    while (1) {
        int nfds = epoll_wait(reactor->epfd, reactor->events, reactor->max_events, -1);
        if (nfds == -1) {
            perror("epoll_wait failed");
            break;
        }

        for (int i = 0; i < nfds; i++) {
            int fd = reactor->events[i].data.fd;
            int events = reactor->events[i].events;

            // 查找对应的事件处理函数
            reactor_event_t *curr = reactor->event_list;
            while (curr) {
                if (curr->fd == fd) {
                    curr->handler(fd, events, curr->arg);
                    break;
                }
                curr = curr->next;
            }
        }
    }
}

void reactor_destroy(reactor_t *reactor) {
    // 清理所有注册的事件
    reactor_event_t *curr = reactor->event_list;
    while (curr) {
        reactor_event_t *temp = curr;
        curr = curr->next;
        close(temp->fd);
        free(temp);
    }

    close(reactor->epfd);
    free(reactor->events);
    free(reactor);
}

3. react_server.c（回显服务器实现） #

#include "reactor.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <signal.h>

#define PORT 8080
#define MAX_EVENTS 10

reactor_t *g_reactor = NULL;

// 信号处理函数：捕获Ctrl+C，优雅退出服务器
void sigint_handler(int sig) {
    printf("\nServer is shutting down...\n");
    if (g_reactor) {
        reactor_destroy(g_reactor);
    }
    exit(0);
}

// 客户端回显处理函数：读取请求并原样返回
void echo_handler(int client_fd, int events, void *arg) {
    char buf[1024];
    ssize_t nread = read(client_fd, buf, sizeof(buf)-1);
    if (nread == -1) {
        perror("read from client failed");
        reactor_remove_event(g_reactor, client_fd);
        return;
    } else if (nread == 0) {
        printf("Client %d disconnected\n", client_fd);
        reactor_remove_event(g_reactor, client_fd);
        return;
    }

    buf[nread] = '\0';
    printf("Received from client %d: %s", client_fd, buf);

    // 原样写回客户端
    ssize_t nwrite = write(client_fd, buf, nread);
    if (nwrite == -1) {
        perror("write to client failed");
        reactor_remove_event(g_reactor, client_fd);
        return;
    }
}

// 监听套接字处理函数：接受客户端连接
void accept_handler(int listen_fd, int events, void *arg) {
    struct sockaddr_in client_addr;
    socklen_t addr_len = sizeof(client_addr);
    int client_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &addr_len);
    if (client_fd == -1) {
        perror("accept failed");
        return;
    }

    printf("New client connected: fd=%d\n", client_fd);

    // 将客户端套接字添加到反应堆，监听读事件
    if (reactor_add_event(g_reactor, client_fd, EPOLLIN | EPOLLET, echo_handler, NULL) == -1) {
        close(client_fd);
        printf("Failed to add client %d to reactor\n", client_fd);
    }
}

int main() {
    // 注册SIGINT信号处理
    struct sigaction sa;
    memset(&sa, 0, sizeof(sa));
    sa.sa_handler = sigint_handler;
    sigaction(SIGINT, &sa, NULL);

    // 创建监听套接字
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd == -1) {
        perror("socket create failed");
        exit(1);
    }

    // 设置端口复用
    int opt = 1;
    if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)) == -1) {
        perror("setsockopt failed");
        close(listen_fd);
        exit(1);
    }

    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(PORT);

    if (bind(listen_fd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
        perror("bind failed");
        close(listen_fd);
        exit(1);
    }

    if (listen(listen_fd, SOMAXCONN) == -1) {
        perror("listen failed");
        close(listen_fd);
        exit(1);
    }

    printf("Server listening on port %d...\n", PORT);

    // 初始化反应堆
    g_reactor = reactor_init(MAX_EVENTS);
    if (!g_reactor) {
        close(listen_fd);
        exit(1);
    }

    // 将监听套接字添加到反应堆，监听读事件
    if (reactor_add_event(g_reactor, listen_fd, EPOLLIN | EPOLLET, accept_handler, NULL) == -1) {
        reactor_destroy(g_reactor);
        close(listen_fd);
        exit(1);
    }

    // 启动反应堆事件循环
    reactor_run(g_reactor);

    return 0;
}

4. Makefile（编译脚本） #

CC = gcc
CFLAGS = -Wall -Wextra -O2

all: react_server

react_server: react_server.o reactor.o
	$(CC) $(CFLAGS) -o $@ $^

react_server.o: react_server.c reactor.h
	$(CC) $(CFLAGS) -c $<

reactor.o: reactor.c reactor.h
	$(CC) $(CFLAGS) -c $<

clean:
	rm -f react_server *.o

关键实现说明 #

• 连接状态反馈：在accept_handler中打印新客户端连接信息，客户端正常断开时在echo_handler中输出断开提示，服务器启动时打印监听端口信息。
• 回显逻辑：echo_handler读取客户端数据后直接原样写回，处理了读失败、客户端主动断开（read返回0）的场景。
• 信号处理：注册SIGINT信号捕获函数，按下Ctrl+C时优雅销毁反应堆、释放资源后退出，避免内存泄漏。
• 高效事件处理：使用EPOLLET边缘触发模式，提升事件处理效率，确保每个就绪事件只被触发一次。
• 资源管理：移除事件时自动关闭文件描述符并释放链表节点，销毁反应堆时清理所有注册资源。

编译与运行步骤 #

1. 执行make编译生成可执行文件react_server
2. 启动服务器：./react_server
3. 测试：使用telnet localhost 8080或nc localhost 8080连接服务器，输入内容后会得到原样回显；客户端按Ctrl+C可断开连接，服务器会打印断开提示；服务器终端按Ctrl+C可优雅关闭服务。

内容的提问来源于stack exchange，提问作者M Amir Khan