（七）内核那些事儿：操作系统对网络包的处理

// 典型的 TCP 服务器端编程流程
int server_fd = socket(AF_INET, SOCK_STREAM, 0);        // 创建 socket
bind(server_fd, &server_addr, sizeof(server_addr));     // 绑定地址和端口
listen(server_fd, SOMAXCONN);                           // 监听连接
int client_fd = accept(server_fd, &client_addr, &len);   // 接受连接
ssize_t bytes = recv(client_fd, buffer, sizeof(buffer), 0); // 接收数据
send(client_fd, response, strlen(response), 0);         // 发送数据
close(client_fd);                                       // 关闭连接

// Linux 内核中的 sk_buff 结构（简化版）
struct sk_buff {
    struct sk_buff *next, *prev;     // 链表指针
    struct net_device *dev;          // 关联的网络设备
    unsigned char *head, *data;      // 数据指针
    unsigned char *tail, *end;       // 缓冲区边界
    unsigned int len, data_len;      // 数据长度
    // ...更多字段
};

fd_set readfds, writefds;
FD_ZERO(&readfds);
FD_SET(sockfd, &readfds);
int ready = select(sockfd + 1, &readfds, &writefds, NULL, &timeout);

struct pollfd fds[MAX_CLIENTS];
fds[0].fd = server_fd;
fds[0].events = POLLIN;
int ready = poll(fds, nfds, timeout);

int epfd = epoll_create1(0);
struct epoll_event ev, events[MAX_EVENTS];
ev.events = EPOLLIN;
ev.data.fd = sockfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);
int nfds = epoll_wait(epfd, events, MAX_EVENTS, timeout);

// 用户程序通过 recv() 系统调用读取数据
ssize_t sock_recvmsg(struct socket *sock, struct msghdr *msg, size_t size, int flags)
{
    struct sock *sk = sock->sk;
    int addr_len = 0;
    int err;

    // 从 socket 接收队列复制数据到用户缓冲区
    err = sk->sk_prot->recvmsg(sk, msg, size, flags, &addr_len);
    if (err >= 0)
        msg->msg_namelen = addr_len;
    return err;
}

// 用户程序发送数据
char *message = "Hello, Network!";
ssize_t sent = send(sockfd, message, strlen(message), 0);

// sys_sendto 系统调用
SYSCALL_DEFINE6(sendto, int, fd, void __user *, buff, size_t, len,
                unsigned int, flags, struct sockaddr __user *, addr,
                int, addr_len)
{
    struct socket *sock;
    struct sockaddr_storage address;
    int err;

    // 获取 socket 对象
    sock = sockfd_lookup_light(fd, &err, &fput_needed);
    if (!sock)
        goto out;

    // 调用协议族的发送函数
    err = sock_sendmsg(sock, &msg, len);

    sockfd_put_light(sock, fput_needed);
out:
    return err;
}

// TCP 发送处理
int tcp_sendmsg(struct sock *sk, struct msghdr *msg, size_t size)
{
    struct tcp_sock *tp = tcp_sk(sk);
    struct sk_buff *skb;
    int flags, err;

    // 检查连接状态
    if (sk->sk_state != TCP_ESTABLISHED)
        return -EPIPE;

    // 分配 sk_buff
    skb = sk_stream_alloc_skb(sk, select_size(sk), sk->sk_allocation);
    if (!skb)
        return -ENOMEM;

    // 复制用户数据到内核缓冲区
    if (skb_add_data(skb, from, copy) != 0) {
        __kfree_skb(skb);
        return -EFAULT;
    }

    // 添加到发送队列
    tcp_push(sk, flags, mss_now, tp->nonagle);

    return size;
}

// IP 层发送
int ip_queue_xmit(struct sk_buff *skb, struct flowi *fl)
{
    struct sock *sk = skb->sk;
    struct rtable *rt;
    struct iphdr *iph;

    // 路由查找
    rt = ip_route_output_flow(&init_net, fl, sk);
    if (IS_ERR(rt))
        goto no_route;

    // 构建 IP 头部
    skb_push(skb, sizeof(struct iphdr));
    iph = ip_hdr(skb);
    iph->version = 4;
    iph->ihl = 5;
    iph->tos = 0;
    iph->tot_len = htons(skb->len);
    iph->id = htons(ip_ident++);
    iph->frag_off = 0;
    iph->ttl = 64;
    iph->protocol = IPPROTO_TCP;
    iph->saddr = fl->fl4_src;
    iph->daddr = fl->fl4_dst;
    iph->check = 0;
    iph->check = ip_fast_csum((unsigned char *)iph, iph->ihl);

    // 传递给网络设备层
    return dst_output(skb);
}

// 网络设备发送
int dev_queue_xmit(struct sk_buff *skb)
{
    struct net_device *dev = skb->dev;
    struct netdev_queue *txq;

    // 选择发送队列
    txq = netdev_pick_tx(dev, skb);

    // 检查设备状态
    if (unlikely(dev->flags & IFF_UP) == 0) {
        kfree_skb(skb);
        return -ENETDOWN;
    }

    // 调用驱动程序发送函数
    return dev_hard_start_xmit(skb, dev, txq);
}

// sendfile() 系统调用 - 直接在内核空间传输文件
ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);

// splice() 系统调用 - 在两个文件描述符间传输数据
ssize_t splice(int fd_in, loff_t *off_in, int fd_out, loff_t *off_out,
               size_t len, unsigned int flags);

// 内存映射 + write
void *mapped = mmap(NULL, file_size, PROT_READ, MAP_PRIVATE, fd, 0);
write(sockfd, mapped, file_size);
munmap(mapped, file_size);

// eBPF 程序示例：丢弃特定端口的包
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/tcp.h>

SEC("prog")
int packet_filter(struct __sk_buff *skb)
{
    void *data = (void *)(long)skb->data;
    void *data_end = (void *)(long)skb->data_end;

    struct ethhdr *eth = data;
    if ((void *)eth + sizeof(*eth) > data_end)
        return XDP_PASS;

    if (eth->h_proto != htons(ETH_P_IP))
        return XDP_PASS;

    struct iphdr *ip = (void *)eth + sizeof(*eth);
    if ((void *)ip + sizeof(*ip) > data_end)
        return XDP_PASS;

    if (ip->protocol != IPPROTO_TCP)
        return XDP_PASS;

    struct tcphdr *tcp = (void *)ip + sizeof(*ip);
    if ((void *)tcp + sizeof(*tcp) > data_end)
        return XDP_PASS;

    // 丢弃目标端口为 8080 的包
    if (ntohs(tcp->dest) == 8080)
        return XDP_DROP;

    return XDP_PASS;
}

// XDP 程序：简单的负载均衡
SEC("xdp")
int load_balancer(struct xdp_md *ctx)
{
    void *data = (void *)(long)ctx->data;
    void *data_end = (void *)(long)ctx->data_end;

    struct ethhdr *eth = data;
    struct iphdr *ip = (void *)eth + sizeof(*eth);

    // 边界检查
    if ((void *)ip + sizeof(*ip) > data_end)
        return XDP_PASS;

    // 基于源 IP 的简单哈希负载均衡
    __u32 hash = ip->saddr % 4;

    switch (hash) {
        case 0:
            // 转发到服务器 1
            return bpf_redirect_map(&server_map, 0, 0);
        case 1:
            // 转发到服务器 2
            return bpf_redirect_map(&server_map, 1, 0);
        default:
            return XDP_PASS;
    }
}

// DPDK 应用程序示例
#include <rte_eal.h>
#include <rte_ethdev.h>
#include <rte_mbuf.h>

#define RX_RING_SIZE 1024
#define TX_RING_SIZE 1024
#define NUM_MBUFS 8191
#define MBUF_CACHE_SIZE 250

int main(int argc, char *argv[])
{
    struct rte_mempool *mbuf_pool;
    uint16_t portid = 0;

    // 初始化 EAL
    int ret = rte_eal_init(argc, argv);
    if (ret < 0)
        rte_exit(EXIT_FAILURE, "Error with EAL initialization\n");

    // 创建内存池
    mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL", NUM_MBUFS,
        MBUF_CACHE_SIZE, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());

    // 配置网络端口
    if (port_init(portid, mbuf_pool) != 0)
        rte_exit(EXIT_FAILURE, "Cannot init port %"PRIu16 "\n", portid);

    // 主循环：接收和处理包
    for (;;) {
        struct rte_mbuf *bufs[BURST_SIZE];
        const uint16_t nb_rx = rte_eth_rx_burst(portid, 0, bufs, BURST_SIZE);

        if (unlikely(nb_rx == 0))
            continue;

        // 处理接收到的包
        for (int i = 0; i < nb_rx; i++) {
            process_packet(bufs[i]);
        }

        // 发送处理后的包
        const uint16_t nb_tx = rte_eth_tx_burst(portid, 0, bufs, nb_rx);

        // 释放未发送的包
        if (unlikely(nb_tx < nb_rx)) {
            for (int i = nb_tx; i < nb_rx; i++)
                rte_pktmbuf_free(bufs[i]);
        }
    }

    return 0;
}

// 使用 epoll 的高性能服务器示例
#include <sys/epoll.h>
#include <unistd.h>

#define MAX_EVENTS 10000
#define BUFFER_SIZE 4096

int main() {
    int server_fd, epoll_fd;
    struct epoll_event ev, events[MAX_EVENTS];
    struct sockaddr_in address;

    // 创建服务器 socket
    server_fd = socket(AF_INET, SOCK_STREAM, 0);

    // 设置非阻塞
    int flags = fcntl(server_fd, F_GETFL, 0);
    fcntl(server_fd, F_SETFL, flags | O_NONBLOCK);

    // 绑定地址
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(8080);
    bind(server_fd, (struct sockaddr *)&address, sizeof(address));

    // 监听连接
    listen(server_fd, SOMAXCONN);

    // 创建 epoll 实例
    epoll_fd = epoll_create1(0);

    // 添加服务器 socket 到 epoll
    ev.events = EPOLLIN;
    ev.data.fd = server_fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &ev);

    // 主事件循环
    for (;;) {
        int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);

        for (int n = 0; n < nfds; ++n) {
            if (events[n].data.fd == server_fd) {
                // 新连接
                int client_fd = accept(server_fd, NULL, NULL);
                if (client_fd != -1) {
                    // 设置非阻塞
                    int flags = fcntl(client_fd, F_GETFL, 0);
                    fcntl(client_fd, F_SETFL, flags | O_NONBLOCK);

                    // 添加到 epoll
                    ev.events = EPOLLIN | EPOLLET;  // 边缘触发
                    ev.data.fd = client_fd;
                    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &ev);
                }
            } else {
                // 客户端数据
                int client_fd = events[n].data.fd;
                char buffer[BUFFER_SIZE];

                ssize_t bytes = read(client_fd, buffer, sizeof(buffer));
                if (bytes > 0) {
                    // 处理请求并发送响应
                    handle_request(client_fd, buffer, bytes);
                } else if (bytes == 0 || errno != EAGAIN) {
                    // 连接关闭或错误
                    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client_fd, NULL);
                    close(client_fd);
                }
            }
        }
    }

    return 0;
}

// 使用原始套接字进行网络监控
#include <netinet/ip.h>
#include <netinet/tcp.h>
#include <netinet/if_ether.h>

int main() {
    int sock_fd;
    char buffer[65536];
    struct sockaddr saddr;
    int saddr_len = sizeof(saddr);

    // 创建原始套接字
    sock_fd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL));
    if (sock_fd < 0) {
        perror("Socket Error");
        return 1;
    }

    while (1) {
        // 接收数据包
        ssize_t data_size = recvfrom(sock_fd, buffer, 65536, 0, &saddr, (socklen_t*)&saddr_len);
        if (data_size < 0) {
            perror("Recvfrom error");
            return 1;
        }

        // 解析以太网头部
        struct ethhdr *eth = (struct ethhdr *)buffer;

        if (ntohs(eth->h_proto) == ETH_P_IP) {
            // 解析 IP 头部
            struct iphdr *iph = (struct iphdr*)(buffer + sizeof(struct ethhdr));

            printf("IP Packet: %s -> ", inet_ntoa(*(struct in_addr*)&iph->saddr));
            printf("%s\n", inet_ntoa(*(struct in_addr*)&iph->daddr));

            if (iph->protocol == IPPROTO_TCP) {
                // 解析 TCP 头部
                struct tcphdr *tcph = (struct tcphdr*)(buffer + sizeof(struct ethhdr) + iph->ihl*4);
                printf("TCP Packet: Port %d -> %d\n", ntohs(tcph->source), ntohs(tcph->dest));

                // 进行流量统计、异常检测等
                analyze_tcp_packet(iph, tcph);
            }
        }
    }

    close(sock_fd);
    return 0;
}

// 简单的 SYN flood 防护
#include <linux/netfilter.h>
#include <linux/netfilter_ipv4.h>

static struct {
    __u32 src_ip;
    unsigned long timestamp;
    int count;
} syn_table[SYN_TABLE_SIZE];

static unsigned int syn_flood_hook(void *priv,
                                   struct sk_buff *skb,
                                   const struct nf_hook_state *state)
{
    struct iphdr *iph = ip_hdr(skb);
    struct tcphdr *tcph;

    if (iph->protocol != IPPROTO_TCP)
        return NF_ACCEPT;

    tcph = tcp_hdr(skb);

    // 检查是否为 SYN 包
    if (tcph->syn && !tcph->ack) {
        int hash = iph->saddr % SYN_TABLE_SIZE;
        unsigned long now = jiffies;

        // 检查频率
        if (syn_table[hash].src_ip == iph->saddr) {
            if (time_before(now, syn_table[hash].timestamp + HZ)) {
                syn_table[hash].count++;

                // 超过阈值，丢弃包
                if (syn_table[hash].count > SYN_THRESHOLD) {
                    printk(KERN_INFO "SYN flood detected from %pI4\n", &iph->saddr);
                    return NF_DROP;
                }
            } else {
                // 重置计数器
                syn_table[hash].count = 1;
                syn_table[hash].timestamp = now;
            }
        } else {
            // 新的源 IP
            syn_table[hash].src_ip = iph->saddr;
            syn_table[hash].count = 1;
            syn_table[hash].timestamp = now;
        }
    }

    return NF_ACCEPT;
}

// 高效的网络编程建议

// 1. 使用合适的 I/O 多路复用机制
#ifdef __linux__
    // Linux 系统优先使用 epoll
    int epfd = epoll_create1(EPOLL_CLOEXEC);
#elif defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__APPLE__)
    // BSD 系统使用 kqueue
    int kq = kqueue();
#else
    // 其他系统降级到 poll
    struct pollfd fds[MAX_FDS];
#endif

// 2. 正确设置 socket 选项
int enable = 1;
setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &enable, sizeof(enable));
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &enable, sizeof(enable));

// 3. 使用适当的缓冲区大小
int buf_size = 1024 * 1024;  // 1MB
setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof(buf_size));
setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof(buf_size));

// 4. 错误处理要完善
ssize_t bytes = recv(sockfd, buffer, sizeof(buffer), 0);
if (bytes < 0) {
    if (errno == EAGAIN || errno == EWOULDBLOCK) {
        // 非阻塞 socket，稍后重试
        continue;
    } else {
        // 真正的错误
        perror("recv failed");
        close(sockfd);
    }
} else if (bytes == 0) {
    // 连接已关闭
    close(sockfd);
}